




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


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

UpperCAmelCase_									=	'\n    Examples:\n        ```py\n        >>> import torch\n        >>> import numpy as np\n\n        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline\n        >>> from transformers import pipeline\n        >>> from diffusers.utils import load_image\n\n\n        >>> def make_hint(image, depth_estimator):\n        ...     image = depth_estimator(image)["depth"]\n        ...     image = np.array(image)\n        ...     image = image[:, :, None]\n        ...     image = np.concatenate([image, image, image], axis=2)\n        ...     detected_map = torch.from_numpy(image).float() / 255.0\n        ...     hint = detected_map.permute(2, 0, 1)\n        ...     return hint\n\n\n        >>> depth_estimator = pipeline("depth-estimation")\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior = pipe_prior.to("cuda")\n\n        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16\n        ... )\n        >>> pipe = pipe.to("cuda")\n\n\n        >>> img = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/cat.png"\n        ... ).resize((768, 768))\n\n        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")\n\n        >>> prompt = "A robot, 4k photo"\n        >>> 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"\n\n        >>> generator = torch.Generator(device="cuda").manual_seed(43)\n\n        >>> image_emb, zero_image_emb = pipe_prior(\n        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator\n        ... ).to_tuple()\n\n        >>> images = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     hint=hint,\n        ...     num_inference_steps=50,\n        ...     generator=generator,\n        ...     height=768,\n        ...     width=768,\n        ... ).images\n\n        >>> images[0].save("robot_cat.png")\n        ```\n'




def 					lowerCamelCase__       (						A__       :					Optional[int]  ,			A__       :					Dict  ,			A__       :					Optional[int]=8	):





       '''simple docstring'''



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





class     lowerCamelCase__(					__lowerCamelCase):
      def __init__(				self: List[Any]    ,       UpperCamelCase_: UNetaDConditionModel    ,       UpperCamelCase_: DDPMScheduler    ,       UpperCamelCase_: VQModel    ,       ):
             super().__init__()

             self.register_modules(
                 unet=UpperCamelCase_    ,       scheduler=UpperCamelCase_    ,       movq=UpperCamelCase_    ,       )
             __lowerCamelCase   =						2 ** (len(self.movq.config.block_out_channels   ) - 1)
      def 			lowerCAmelCase__						(				self: int    ,       UpperCamelCase_: Any    ,       UpperCamelCase_: Tuple    ,       UpperCamelCase_: int    ,       UpperCamelCase_: Dict    ,       UpperCamelCase_: Dict    ,       UpperCamelCase_: int   ):
             if latents is None:
                    __lowerCamelCase   =						randn_tensor(UpperCamelCase_    ,       generator=UpperCamelCase_    ,       device=UpperCamelCase_    ,       dtype=UpperCamelCase_   )
             else:
                    if latents.shape != shape:
                           raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {shape}'   )
                    __lowerCamelCase   =						latents.to(UpperCamelCase_   )

             __lowerCamelCase   =						latents * scheduler.init_noise_sigma
             return latents
      def 			lowerCAmelCase__						(				self: Dict    ,       UpperCamelCase_: str=0   ):
             if is_accelerate_available():
                    from accelerate import cpu_offload
             else:
                    raise ImportError("""Please install accelerate via `pip install accelerate`"""   )

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

             __lowerCamelCase   =						[
                 self.unet,
                 self.movq,
             ]
             for cpu_offloaded_model in models:
                    if cpu_offloaded_model is not None:
                           cpu_offload(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: List[str]    ,       UpperCamelCase_: Optional[int]=0   ):
             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."""   )

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

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

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

             # We'll offload the last model manually.
             __lowerCamelCase   =						hook
      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def 			lowerCAmelCase__						(				self: int   ):
             if not hasattr(self.unet    ,       """_hf_hook"""   ):
                    return self.device
             for module in self.unet.modules():
                    if (
                        hasattr(UpperCamelCase_    ,       """_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(UpperCamelCase_   )
      def __call__(				self: Tuple    ,       UpperCamelCase_: Union[torch.FloatTensor, List[torch.FloatTensor]]    ,       UpperCamelCase_: Union[torch.FloatTensor, List[torch.FloatTensor]]    ,       UpperCamelCase_: torch.FloatTensor    ,       UpperCamelCase_: int = 5_12    ,       UpperCamelCase_: int = 5_12    ,       UpperCamelCase_: int = 1_00    ,       UpperCamelCase_: float = 4.0    ,       UpperCamelCase_: int = 1    ,       UpperCamelCase_: Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,       UpperCamelCase_: Optional[torch.FloatTensor] = None    ,       UpperCamelCase_: Optional[str] = "pil"    ,       UpperCamelCase_: bool = True    ,       ):
             __lowerCamelCase   =						self._execution_device

             __lowerCamelCase   =						guidance_scale > 1.0

             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )
             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )
             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )

             __lowerCamelCase   =						image_embeds.shape[0] * num_images_per_prompt

             if do_classifier_free_guidance:
                    __lowerCamelCase   =						image_embeds.repeat_interleave(UpperCamelCase_    ,       dim=0   )
                    __lowerCamelCase   =						negative_image_embeds.repeat_interleave(UpperCamelCase_    ,       dim=0   )
                    __lowerCamelCase   =						hint.repeat_interleave(UpperCamelCase_    ,       dim=0   )

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

             __lowerCamelCase   =						self.movq.config.latent_channels

             __lowerCamelCase,   __lowerCamelCase   =						downscale_height_and_width(UpperCamelCase_    ,       UpperCamelCase_    ,       self.movq_scale_factor   )

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

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

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

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

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

             if output_type == "pil":
                    __lowerCamelCase   =						self.numpy_to_pil(UpperCamelCase_   )

             if not return_dict:
                    return (image,)

             return ImagePipelineOutput(images=UpperCamelCase_   )







import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / 'utils'))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


UpperCAmelCase_									=	get_tests_dir('fixtures')
UpperCAmelCase_									=	get_tests_dir('fixtures/dummy_feature_extractor_config.json')
UpperCAmelCase_									=	get_tests_dir('fixtures/dummy-config.json')





class     lowerCamelCase__(					unittest.TestCase):
      def 			lowerCAmelCase__						(				self: List[str]   ):
             __lowerCamelCase   =						0
      def 			lowerCAmelCase__						(				self: List[str]   ):
             __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""   )
             self.assertIsInstance(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: Union[str, Any]   ):
             __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_   )
             self.assertIsInstance(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: int   ):
             with tempfile.TemporaryDirectory() as tmpdirname:
                    __lowerCamelCase   =						WavaVecaConfig()

                    # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_   ).to_dict()

                    config_dict.pop("""feature_extractor_type"""   )
                    __lowerCamelCase   =						WavaVecaFeatureExtractor(**UpperCamelCase_   )

                    # save in new folder
                    model_config.save_pretrained(UpperCamelCase_   )
                    config.save_pretrained(UpperCamelCase_   )

                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_   )

                    # make sure private variable is not incorrectly saved
                    __lowerCamelCase   =						json.loads(config.to_json_string()   )
                    self.assertTrue("""_processor_class""" not in dict_as_saved   )

             self.assertIsInstance(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: Tuple   ):
             __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_   )
             self.assertIsInstance(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: int   ):
             with self.assertRaisesRegex(
                 UpperCamelCase_    ,       """bert-base is not a local folder and is not a valid model identifier"""   ):
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained("""bert-base"""   )
      def 			lowerCAmelCase__						(				self: Tuple   ):
             with self.assertRaisesRegex(
                 UpperCamelCase_    ,       r"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""   ):
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_    ,       revision="""aaaaaa"""   )
      def 			lowerCAmelCase__						(				self: Optional[Any]   ):
             with self.assertRaisesRegex(
                 UpperCamelCase_    ,       """hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""    ,       ):
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""   )
      def 			lowerCAmelCase__						(				self: Tuple   ):
             # If remote code is not set, we will time out when asking whether to load the model.
             with self.assertRaises(UpperCamelCase_   ):
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                        """hf-internal-testing/test_dynamic_feature_extractor"""   )
             # If remote code is disabled, we can't load this config.
             with self.assertRaises(UpperCamelCase_   ):
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                        """hf-internal-testing/test_dynamic_feature_extractor"""    ,       trust_remote_code=UpperCamelCase_   )

             __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                 """hf-internal-testing/test_dynamic_feature_extractor"""    ,       trust_remote_code=UpperCamelCase_   )
             self.assertEqual(feature_extractor.__class__.__name__    ,       """NewFeatureExtractor"""   )

             # Test feature extractor can be reloaded.
             with tempfile.TemporaryDirectory() as tmp_dir:
                    feature_extractor.save_pretrained(UpperCamelCase_   )
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_    ,       trust_remote_code=UpperCamelCase_   )
             self.assertEqual(reloaded_feature_extractor.__class__.__name__    ,       """NewFeatureExtractor"""   )
      def 			lowerCAmelCase__						(				self: Any   ):
             try:
                    AutoConfig.register("""custom"""    ,       UpperCamelCase_   )
                    AutoFeatureExtractor.register(UpperCamelCase_    ,       UpperCamelCase_   )
                    # Trying to register something existing in the Transformers library will raise an error
                    with self.assertRaises(UpperCamelCase_   ):
                           AutoFeatureExtractor.register(UpperCamelCase_    ,       UpperCamelCase_   )

                    # Now that the config is registered, it can be used as any other config with the auto-API
                    __lowerCamelCase   =						CustomFeatureExtractor.from_pretrained(UpperCamelCase_   )
                    with tempfile.TemporaryDirectory() as tmp_dir:
                           feature_extractor.save_pretrained(UpperCamelCase_   )
                           __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(UpperCamelCase_   )
                           self.assertIsInstance(UpperCamelCase_    ,       UpperCamelCase_   )

             finally:
                    if "custom" in CONFIG_MAPPING._extra_content:
                           del CONFIG_MAPPING._extra_content["custom"]
                    if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                           del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]




      def 			lowerCAmelCase__						(				self: Dict   ):





             class     lowerCamelCase__(					__lowerCamelCase):
                   UpperCAmelCase__      :	str    		=   True

             try:
                    AutoConfig.register("""custom"""    ,       UpperCamelCase_   )
                    AutoFeatureExtractor.register(UpperCamelCase_    ,       UpperCamelCase_   )
                    # If remote code is not set, the default is to use local
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                        """hf-internal-testing/test_dynamic_feature_extractor"""   )
                    self.assertEqual(feature_extractor.__class__.__name__    ,       """NewFeatureExtractor"""   )
                    self.assertTrue(feature_extractor.is_local   )

                    # If remote code is disabled, we load the local one.
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                        """hf-internal-testing/test_dynamic_feature_extractor"""    ,       trust_remote_code=UpperCamelCase_   )
                    self.assertEqual(feature_extractor.__class__.__name__    ,       """NewFeatureExtractor"""   )
                    self.assertTrue(feature_extractor.is_local   )

                    # If remote is enabled, we load from the Hub
                    __lowerCamelCase   =						AutoFeatureExtractor.from_pretrained(
                        """hf-internal-testing/test_dynamic_feature_extractor"""    ,       trust_remote_code=UpperCamelCase_   )
                    self.assertEqual(feature_extractor.__class__.__name__    ,       """NewFeatureExtractor"""   )
                    self.assertTrue(not hasattr(UpperCamelCase_    ,       """is_local"""   )   )

             finally:
                    if "custom" in CONFIG_MAPPING._extra_content:
                           del CONFIG_MAPPING._extra_content["custom"]
                    if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                           del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]

