




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    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__(					__lowerCamelCase ,     unittest.TestCase):
      UpperCAmelCase__      :	Dict    		=   KandinskyVaaImgaImgPipeline
      UpperCAmelCase__      :	Optional[int]    		=   ['image_embeds', 'negative_image_embeds', 'image']
      UpperCAmelCase__      :	Optional[int]    		=   [
          'image_embeds',
          'negative_image_embeds',
          'image',
      ]
      UpperCAmelCase__      :	Optional[Any]    		=   [
          'generator',
          'height',
          'width',
          'strength',
          'guidance_scale',
          'num_inference_steps',
          'return_dict',
          'guidance_scale',
          'num_images_per_prompt',
          'output_type',
          'return_dict',
      ]
      UpperCAmelCase__      :	Optional[Any]    		=   False
      @property
      def 			lowerCAmelCase__						(				self: str   ):
             return 32
      @property
      def 			lowerCAmelCase__						(				self: Tuple   ):
             return 32
      @property
      def 			lowerCAmelCase__						(				self: int   ):
             return self.time_input_dim
      @property
      def 			lowerCAmelCase__						(				self: int   ):
             return self.time_input_dim * 4
      @property
      def 			lowerCAmelCase__						(				self: Tuple   ):
             return 1_00
      @property
      def 			lowerCAmelCase__						(				self: Optional[Any]   ):
             torch.manual_seed(0   )

             __lowerCamelCase   =						{
                 """in_channels""": 4,
                 # 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,
             }

             __lowerCamelCase   =						UNetaDConditionModel(**UpperCamelCase_   )
             return model
      @property
      def 			lowerCAmelCase__						(				self: Any   ):
             return {
                 "block_out_channels": [32, 64],
                 "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": 12,
                 "out_channels": 3,
                 "up_block_types": [
                     "AttnUpDecoderBlock2D",
                     "UpDecoderBlock2D",
                 ],
                 "vq_embed_dim": 4,
             }
      @property
      def 			lowerCAmelCase__						(				self: int   ):
             torch.manual_seed(0   )
             __lowerCamelCase   =						VQModel(**self.dummy_movq_kwargs   )
             return model
      def 			lowerCAmelCase__						(				self: Optional[Any]   ):
             __lowerCamelCase   =						self.dummy_unet
             __lowerCamelCase   =						self.dummy_movq

             __lowerCamelCase   =						{
                 """num_train_timesteps""": 10_00,
                 """beta_schedule""": """linear""",
                 """beta_start""": 0.0_0085,
                 """beta_end""": 0.012,
                 """clip_sample""": False,
                 """set_alpha_to_one""": False,
                 """steps_offset""": 0,
                 """prediction_type""": """epsilon""",
                 """thresholding""": False,
             }

             __lowerCamelCase   =						DDIMScheduler(**UpperCamelCase_   )

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

             return components
      def 			lowerCAmelCase__						(				self: Dict    ,       UpperCamelCase_: str    ,       UpperCamelCase_: Optional[int]=0   ):
             __lowerCamelCase   =						floats_tensor((1, self.text_embedder_hidden_size)    ,       rng=random.Random(UpperCamelCase_   )   ).to(UpperCamelCase_   )
             __lowerCamelCase   =						floats_tensor((1, self.text_embedder_hidden_size)    ,       rng=random.Random(seed + 1   )   ).to(
                 UpperCamelCase_   )
             # create init_image
             __lowerCamelCase   =						floats_tensor((1, 3, 64, 64)    ,       rng=random.Random(UpperCamelCase_   )   ).to(UpperCamelCase_   )
             __lowerCamelCase   =						image.cpu().permute(0    ,       2    ,       3    ,       1   )[0]
             __lowerCamelCase   =						Image.fromarray(np.uinta(UpperCamelCase_   )   ).convert("""RGB"""   ).resize((2_56, 2_56)   )

             if str(UpperCamelCase_   ).startswith("""mps"""   ):
                    __lowerCamelCase   =						torch.manual_seed(UpperCamelCase_   )
             else:
                    __lowerCamelCase   =						torch.Generator(device=UpperCamelCase_   ).manual_seed(UpperCamelCase_   )
             __lowerCamelCase   =						{
                 """image""": init_image,
                 """image_embeds""": image_embeds,
                 """negative_image_embeds""": negative_image_embeds,
                 """generator""": generator,
                 """height""": 64,
                 """width""": 64,
                 """num_inference_steps""": 10,
                 """guidance_scale""": 7.0,
                 """strength""": 0.2,
                 """output_type""": """np""",
             }
             return inputs




      def 			lowerCAmelCase__						(				self: Dict   ):
             __lowerCamelCase   =						"""cpu"""

             __lowerCamelCase   =						self.get_dummy_components()

             __lowerCamelCase   =						self.pipeline_class(**UpperCamelCase_   )
             __lowerCamelCase   =						pipe.to(UpperCamelCase_   )

             pipe.set_progress_bar_config(disable=UpperCamelCase_   )

             __lowerCamelCase   =						pipe(**self.get_dummy_inputs(UpperCamelCase_   )   )
             __lowerCamelCase   =						output.images

             __lowerCamelCase   =						pipe(
                 **self.get_dummy_inputs(UpperCamelCase_   )    ,       return_dict=UpperCamelCase_    ,       )[0]

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

             assert image.shape == (1, 64, 64, 3)

             __lowerCamelCase   =						np.array(
                 [0.619_9778, 0.6398_4406, 0.4614_5785, 0.6294_4984, 0.562_2215, 0.4730_6132, 0.4744_1456, 0.460_7606, 0.4871_9263]   )
             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()}'





@slow
@require_torch_gpu
class     lowerCamelCase__(					unittest.TestCase):
      def 			lowerCAmelCase__						(				self: Dict   ):
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()




      def 			lowerCAmelCase__						(				self: Union[str, Any]   ):
             __lowerCamelCase   =						load_numpy(
                 """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                 """/kandinskyv22/kandinskyv22_img2img_frog.npy"""   )

             __lowerCamelCase   =						load_image(
                 """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png"""   )
             __lowerCamelCase   =						"""A red cartoon frog, 4k"""

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

             __lowerCamelCase   =						KandinskyVaaImgaImgPipeline.from_pretrained(
                 """kandinsky-community/kandinsky-2-2-decoder"""    ,       torch_dtype=torch.floataa   )
             __lowerCamelCase   =						pipeline.to(UpperCamelCase_   )

             pipeline.set_progress_bar_config(disable=UpperCamelCase_   )

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

             __lowerCamelCase   =						pipeline(
                 image=UpperCamelCase_    ,       image_embeds=UpperCamelCase_    ,       negative_image_embeds=UpperCamelCase_    ,       generator=UpperCamelCase_    ,       num_inference_steps=1_00    ,       height=7_68    ,       width=7_68    ,       strength=0.2    ,       output_type="""np"""    ,       )

             __lowerCamelCase   =						output.images[0]

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

             assert_mean_pixel_difference(UpperCamelCase_    ,       UpperCamelCase_   )







import json
import os
from datetime import date
from pathlib import Path

from tabulate import DataRow, TableFormat, tabulate


UpperCAmelCase_									=	TableFormat(
    lineabove=None,
    linebelowheader=None,
    linebetweenrows=None,
    linebelow=None,
    headerrow=DataRow('', '|', '|'),
    datarow=DataRow('', '|', '|'),
    padding=1,
    with_header_hide=None,
)


UpperCAmelCase_									=	[]
UpperCAmelCase_									=	[]

UpperCAmelCase_									=	{'type': 'section', 'text': {'type': 'plain_text', 'text': 'No failed tests! 🤗', 'emoji': True}}

UpperCAmelCase_									=	[
    {
        'type': 'header',
        'text': {
            'type': 'plain_text',
            'text': f"""🤗 Accelerate nightly {os.environ.get("TEST_TYPE", "")} test results""",
            'emoji': True,
        },
    }
]

UpperCAmelCase_									=	0
for log in Path().glob('*.log'):
       UpperCAmelCase_									=	0
       with open(log, 'r') as f:
              for line in f:
                     UpperCAmelCase_									=	json.loads(line)
                     if line.get('nodeid', '') != "":
                            UpperCAmelCase_									=	line['nodeid']
                            if line.get('duration', None) is not None:
                                   UpperCAmelCase_									=	f"""{line["duration"]:.4f}"""
                                   if line.get('outcome', '') == "failed":
                                          section_num_failed += 1
                                          failed.append([test, duration, log.name.split('_')[0]])
                                          total_num_failed += 1
       group_info.append([str(log), section_num_failed, failed])
       UpperCAmelCase_									=	[]
       log.unlink()

UpperCAmelCase_									=	''
UpperCAmelCase_									=	[]
if total_num_failed > 0:
       for name, num_failed, failed_tests in group_info:
              if num_failed > 0:
                     if num_failed == 1:
                            message += f"*{name[1:]}: {num_failed} failed test*\n"
                     else:
                            message += f"*{name[1:]}: {num_failed} failed tests*\n"
                     UpperCAmelCase_									=	[]
                     UpperCAmelCase_									=	{}
                     for test in failed_tests:
                            UpperCAmelCase_									=	test[0].split('::')
                            UpperCAmelCase_									=	data[0].split('/')[-1]
                            if data[0] not in filesafailed:
                                   UpperCAmelCase_									=	[data[1:]]
                            else:
                                   filesafailed[data[0]] += [data[1:]]
                            failed_table.append(data)

                     UpperCAmelCase_									=	[test[0] for test in failed_table]
                     UpperCAmelCase_									=	list(set(files))
                     # Count number of instances in failed_tests
                     UpperCAmelCase_									=	[]
                     for file in individual_files:
                            table.append([file, len(filesafailed[file])])

                     UpperCAmelCase_									=	tabulate(
                         table,
                         headers=['Test Location', 'Num Failed'],
                         tablefmt=hf_table_format,
                         stralign='right',
                     )
                     message += f"\n```\n{failed_table}\n```"
                     all_filesafailed.append(filesafailed)
       if len(message) > 3_000:
              UpperCAmelCase_									=	'Too many failed tests, please see the full report in the Action results.'
              UpperCAmelCase_									=	len(err) + 10
              UpperCAmelCase_									=	message[: 3_000 - offset] + f"""\n...\n```\n{err}"""
       print(f"""### {message}""")
else:
       UpperCAmelCase_									=	'No failed tests! 🤗'
       print(f"""## {message}""")
       payload.append(no_error_payload)

if os.environ.get('TEST_TYPE', '') != "":
       from slack_sdk import WebClient

       UpperCAmelCase_									=	WebClient(token=os.environ['SLACK_API_TOKEN'])
       if message != "No failed tests! 🤗":
              UpperCAmelCase_									=	{
                  'type': 'section',
                  'text': {
                      'type': 'mrkdwn',
                      'text': message,
                  },
              }
              payload.append(md_report)
              UpperCAmelCase_									=	{
                  'type': 'section',
                  'text': {
                      'type': 'mrkdwn',
                      'text': '*For more details:*',
                  },
                  'accessory': {
                      'type': 'button',
                      'text': {
                          'type': 'plain_text',
                          'text': 'Check Action results',
                          'emoji': True,
                      },
                      'url': f"""https://github.com/{os.environ["GITHUB_REPOSITORY"]}/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
                  },
              }
              payload.append(action_button)
              UpperCAmelCase_									=	{
                  'type': 'context',
                  'elements': [
                      {
                          'type': 'plain_text',
                          'text': f"""Nightly {os.environ.get("TEST_TYPE")} test results for {date.today()}""",
                      }
                  ],
              }
              payload.append(date_report)
       UpperCAmelCase_									=	client.chat_postMessage(channel='#accelerate-ci-daily', text=message, blocks=payload)
       UpperCAmelCase_									=	response.data['ts']
       for failed_file in all_filesafailed:
              for test_location, test_failures in failed_file.items():
                     # Keep only the first instance of the test name
                     UpperCAmelCase_									=	''
                     for i, row in enumerate(test_failures):
                            if row[0] != test_class:
                                   UpperCAmelCase_									=	row[0]
                            else:
                                   UpperCAmelCase_									=	''

                     UpperCAmelCase_									=	{
                         'type': 'section',
                         'text': {
                             'type': 'mrkdwn',
                             'text': f"""Test location: {test_location}\n```\n{tabulate(test_failures, headers=["Class", "Test"], tablefmt=hf_table_format, stralign="right")}\n```""",
                         },
                     }

                     client.chat_postMessage(
                         channel='#accelerate-ci-daily',
                         thread_ts=ts,
                         blocks=[payload],
                     )

