
import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaInpaintPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class 					lowerCamelCase		(  A_							,			unittest.TestCase  ):
     UpperCAmelCase__    :							List[str]									=		KandinskyVaaInpaintPipeline
     UpperCAmelCase__    :							List[Any]									=		["image_embeds", "negative_image_embeds", "image", "mask_image"]
     UpperCAmelCase__    :							Tuple									=		[
         "image_embeds",
         "negative_image_embeds",
         "image",
         "mask_image",
     ]
     UpperCAmelCase__    :							List[Any]									=		[
         "generator",
         "height",
         "width",
         "latents",
         "guidance_scale",
         "num_inference_steps",
         "return_dict",
         "guidance_scale",
         "num_images_per_prompt",
         "output_type",
         "return_dict",
     ]
     UpperCAmelCase__    :							Any									=		False





     @property
     def 		UpperCAmelCase(self				:   List[str]       )						->						Union[str, Any]:
         return 3_2





     @property
     def 		UpperCAmelCase(self				:   Dict       )						->						Union[str, Any]:
         return 3_2





     @property
     def 		UpperCAmelCase(self				:   List[Any]       )						->						int:
         return self.time_input_dim





     @property
     def 		UpperCAmelCase(self				:   List[str]       )						->						Tuple:
         return self.time_input_dim * 4





     @property
     def 		UpperCAmelCase(self				:   List[str]       )						->						Optional[Any]:
         return 1_0_0





     @property
     def 		UpperCAmelCase(self				:   Optional[Any]       )						->						Union[str, Any]:
         torch.manual_seed(0       )

         snake_case								= {
             "in_channels": 9,
             # Out channels is double in channels because predicts mean and variance
             "out_channels": 8,
             "addition_embed_type": "image",
             "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
             "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
             "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
             "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
             "layers_per_block": 1,
             "encoder_hid_dim": self.text_embedder_hidden_size,
             "encoder_hid_dim_type": "image_proj",
             "cross_attention_dim": self.cross_attention_dim,
             "attention_head_dim": 4,
             "resnet_time_scale_shift": "scale_shift",
             "class_embed_type": None,
         }

         snake_case								= UNetaDConditionModel(**_A       )
         return model





     @property
     def 		UpperCAmelCase(self				:   Tuple       )						->						int:
         return {
             "block_out_channels": [3_2, 6_4],
             "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
             "in_channels": 3,
             "latent_channels": 4,
             "layers_per_block": 1,
             "norm_num_groups": 8,
             "norm_type": "spatial",
             "num_vq_embeddings": 1_2,
             "out_channels": 3,
             "up_block_types": [
                 "AttnUpDecoderBlock2D",
                 "UpDecoderBlock2D",
             ],
             "vq_embed_dim": 4,
         }





     @property
     def 		UpperCAmelCase(self				:   Any       )						->						int:
         torch.manual_seed(0       )
         snake_case								= VQModel(**self.dummy_movq_kwargs       )
         return model





     def 		UpperCAmelCase(self				:   int       )						->						List[str]:
         snake_case								= self.dummy_unet
         snake_case								= self.dummy_movq

         snake_case								= DDIMScheduler(
             num_train_timesteps=1_0_0_0    ,   beta_schedule="linear"    ,   beta_start=0.0_00_85    ,   beta_end=0.0_12    ,   clip_sample=_A    ,   set_alpha_to_one=_A    ,   steps_offset=1    ,   prediction_type="epsilon"    ,   thresholding=_A    ,   )

         snake_case								= {
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components





     def 		UpperCAmelCase(self				:   Tuple    ,   _A				:   int    ,   _A				:   Optional[Any]=0       )						->						Tuple:
         snake_case								= floats_tensor((1, self.text_embedder_hidden_size)    ,   rng=random.Random(_A       )       ).to(_A       )
         snake_case								= floats_tensor((1, self.text_embedder_hidden_size)    ,   rng=random.Random(seed + 1       )       ).to(
             _A       )
         # create init_image
         snake_case								= floats_tensor((1, 3, 6_4, 6_4)    ,   rng=random.Random(_A       )       ).to(_A       )
         snake_case								= image.cpu().permute(0    ,   2    ,   3    ,   1       )[0]
         snake_case								= Image.fromarray(np.uinta(_A       )       ).convert("RGB"       ).resize((2_5_6, 2_5_6)       )
         # create mask
         snake_case								= np.ones((6_4, 6_4)    ,   dtype=np.floataa       )
         snake_case								= 0

         if str(_A       ).startswith("mps"       ):
             snake_case								= torch.manual_seed(_A       )
         else:
             snake_case								= torch.Generator(device=_A       ).manual_seed(_A       )
         snake_case								= {
             "image": init_image,
             "mask_image": mask,
             "image_embeds": image_embeds,
             "negative_image_embeds": negative_image_embeds,
             "generator": generator,
             "height": 6_4,
             "width": 6_4,
             "num_inference_steps": 2,
             "guidance_scale": 4.0,
             "output_type": "np",
         }
         return inputs





     def 		UpperCAmelCase(self				:   Optional[Any]       )						->						Tuple:
         snake_case								= "cpu"

         snake_case								= self.get_dummy_components()

         snake_case								= self.pipeline_class(**_A       )
         snake_case								= pipe.to(_A       )

         pipe.set_progress_bar_config(disable=_A       )

         snake_case								= pipe(**self.get_dummy_inputs(_A       )       )
         snake_case								= output.images

         snake_case								= pipe(
             **self.get_dummy_inputs(_A       )    ,   return_dict=_A    ,   )[0]

         snake_case								= image[0, -3:, -3:, -1]
         snake_case								= image_from_tuple[0, -3:, -3:, -1]

         print(f'image.shape {image.shape}'       )

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case								= np.array(
             [0.50_77_59_03, 0.49_52_71_95, 0.48_82_45_43, 0.50_19_22_37, 0.48_64_49_06, 0.49_37_38_14, 0.4_78_05_98, 0.47_23_48_27, 0.48_32_78_48]       )

         assert (
             np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
         ), f' expected_slice {expected_slice}, but got {image_slice.flatten()}'
         assert (
             np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2
         ), f' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'





     def 		UpperCAmelCase(self				:   Tuple       )						->						Union[str, Any]:
         super().test_inference_batch_single_identical(expected_max_diff=3E-3       )







@slow
@require_torch_gpu
class 					lowerCamelCase		(  unittest.TestCase  ):





     def 		UpperCAmelCase(self				:   Optional[Any]       )						->						Tuple:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





     def 		UpperCAmelCase(self				:   Optional[int]       )						->						Optional[Any]:
         snake_case								= load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/kandinskyv22_inpaint_cat_with_hat_fp16.npy"       )

         snake_case								= load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"       )
         snake_case								= np.ones((7_6_8, 7_6_8)    ,   dtype=np.floataa       )
         snake_case								= 0

         snake_case								= "a hat"

         snake_case								= KandinskyVaaPriorPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-prior"    ,   torch_dtype=torch.floataa       )
         pipe_prior.to(_A       )

         snake_case								= KandinskyVaaInpaintPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-decoder-inpaint"    ,   torch_dtype=torch.floataa       )
         snake_case								= pipeline.to(_A       )
         pipeline.set_progress_bar_config(disable=_A       )

         snake_case								= torch.Generator(device="cpu"       ).manual_seed(0       )
         snake_case					,							snake_case								= pipe_prior(
             _A    ,   generator=_A    ,   num_inference_steps=5    ,   negative_prompt=""    ,   ).to_tuple()

         snake_case								= pipeline(
             image=_A    ,   mask_image=_A    ,   image_embeds=_A    ,   negative_image_embeds=_A    ,   generator=_A    ,   num_inference_steps=1_0_0    ,   height=7_6_8    ,   width=7_6_8    ,   output_type="np"    ,   )

         snake_case								= output.images[0]

         assert image.shape == (7_6_8, 7_6_8, 3)

         assert_mean_pixel_difference(_A    ,   _A       )



from collections import defaultdict







class 					lowerCamelCase		:





     def __init__(self				:   Tuple    ,   _A				:   Optional[int]    ,   _A				:   List[str]       )						->						Union[str, Any]:
         snake_case								= total  # total no of tasks (N)

         # DP table will have a dimension of (2^M)*N
         # initially all values are set to -1
         snake_case								= [
             [-1 for i in range(total + 1       )] for j in range(2 ** len(_A       )       )
         ]

         snake_case								= defaultdict(_A       )  # stores the list of persons for each task

         # final_mask is used to check if all persons are included by setting all bits
         # to 1
         snake_case								= (1 << len(_A       )) - 1





     def 		UpperCAmelCase(self				:   str    ,   _A				:   Optional[Any]    ,   _A				:   List[Any]       )						->						str:
         # if mask == self.finalmask all persons are distributed tasks, return 1
         if mask == self.final_mask:
             return 1

         # if not everyone gets the task and no more tasks are available, return 0
         if task_no > self.total_tasks:
             return 0

         # if case already considered
         if self.dp[mask][task_no] != -1:
             return self.dp[mask][task_no]

         # Number of ways when we don't this task in the arrangement
         snake_case								= self.count_ways_until(_A    ,   task_no + 1       )

         # now assign the tasks one by one to all possible persons and recursively
         # assign for the remaining tasks.
         if task_no in self.task:
             for p in self.task[task_no]:
                 # if p is already given a task
                 if mask & (1 << p):
                     continue

                 # assign this task to p and change the mask value. And recursively
                 # assign tasks with the new mask value.
                 total_ways_util += self.count_ways_until(mask | (1 << p)    ,   task_no + 1       )

        # save the value.
         snake_case								= total_ways_util

         return self.dp[mask][task_no]





     def 		UpperCAmelCase(self				:   Any    ,   _A				:   Dict       )						->						Optional[Any]:
         # Store the list of persons for each task
         for i in range(len(_A       )       ):
             for j in task_performed[i]:
                 self.task[j].append(_A       )

        # call the function to fill the DP table, final answer is stored in dp[0][1]
         return self.count_ways_until(0    ,   1       )


if __name__ == "__main__":
   _A   =			5  # total no of tasks (the value of N)

   # the list of tasks that can be done by M persons.
   _A   =			[[1, 3, 4], [1, 2, 5], [3, 4]]
   print(
       AssignmentUsingBitmask(task_performed, total_tasks).count_no_of_ways(
           task_performed
       )
   )

