


import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    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 					A							(    UpperCAmelCase_		,    unittest.TestCase					):
   __UpperCAmelCase			:			str              =      KandinskyVaaControlnetPipeline
   __UpperCAmelCase			:			Tuple              =      ['image_embeds', 'negative_image_embeds', 'hint']
   __UpperCAmelCase			:			int              =      ['image_embeds', 'negative_image_embeds', 'hint']
   __UpperCAmelCase			:			Optional[int]              =      [
       'generator',
       'height',
       'width',
       'latents',
       'guidance_scale',
       'num_inference_steps',
       'return_dict',
       'guidance_scale',
       'num_images_per_prompt',
       'output_type',
       'return_dict',
   ]
   __UpperCAmelCase			:			List[str]              =      False


   @property
   def       lowercase_				(self					:						int		)						->    List[Any]:
         """simple docstring"""



         return 3_2


   @property
   def       lowercase_				(self					:						str		)						->    Tuple:
         """simple docstring"""



         return 3_2


   @property
   def       lowercase_				(self					:						str		)						->    Any:
         """simple docstring"""



         return self.time_input_dim


   @property
   def       lowercase_				(self					:						Optional[Any]		)						->    List[str]:
         """simple docstring"""



         return self.time_input_dim * 4


   @property
   def       lowercase_				(self					:						Optional[Any]		)						->    List[str]:
         """simple docstring"""



         return 1_0_0


   @property
   def       lowercase_				(self					:						Dict		)						->    Optional[Any]:
         """simple docstring"""



         torch.manual_seed(0		)

         UpperCAmelCase__								=  {
             "in_channels": 8,
             # Out channels is double in channels because predicts mean and variance
             "out_channels": 8,
             "addition_embed_type": "image_hint",
             "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,
         }

         UpperCAmelCase__								=  UNetaDConditionModel(**__UpperCAmelCase		)
         return model


   @property
   def       lowercase_				(self					:						List[Any]		)						->    Tuple:
         """simple docstring"""



         return {
             "block_out_channels": [3_2, 3_2, 6_4, 6_4],
             "down_block_types": [
                 "DownEncoderBlock2D",
                 "DownEncoderBlock2D",
                 "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", "UpDecoderBlock2D", "UpDecoderBlock2D"],
             "vq_embed_dim": 4,
         }


   @property
   def       lowercase_				(self					:						int		)						->    int:
         """simple docstring"""



         torch.manual_seed(0		)
         UpperCAmelCase__								=  VQModel(**self.dummy_movq_kwargs		)
         return model


   def       lowercase_				(self					:						List[Any]		)						->    List[str]:
         """simple docstring"""



         UpperCAmelCase__								=  self.dummy_unet
         UpperCAmelCase__								=  self.dummy_movq

         UpperCAmelCase__								=  DDIMScheduler(
             num_train_timesteps=1_0_0_0	,  beta_schedule="linear"	,  beta_start=0.00085	,  beta_end=0.012	,  clip_sample=__UpperCAmelCase	,  set_alpha_to_one=__UpperCAmelCase	,  steps_offset=1	,  prediction_type="epsilon"	,  thresholding=__UpperCAmelCase	,  )

         UpperCAmelCase__								=  {
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }
         return components


   def       lowercase_				(self					:						Union[str, Any]	,  __UpperCAmelCase					:						Optional[int]	,  __UpperCAmelCase					:						Optional[Any]=0		)						->    List[Any]:
         """simple docstring"""



         UpperCAmelCase__								=  floats_tensor((1, self.text_embedder_hidden_size)	,  rng=random.Random(__UpperCAmelCase		)		).to(__UpperCAmelCase		)
         UpperCAmelCase__								=  floats_tensor((1, self.text_embedder_hidden_size)	,  rng=random.Random(seed + 1		)		).to(
             __UpperCAmelCase		)

         # create hint
         UpperCAmelCase__								=  floats_tensor((1, 3, 6_4, 6_4)	,  rng=random.Random(__UpperCAmelCase		)		).to(__UpperCAmelCase		)

         if str(__UpperCAmelCase		).startswith("mps"		):
               UpperCAmelCase__								=  torch.manual_seed(__UpperCAmelCase		)
         else:
               UpperCAmelCase__								=  torch.Generator(device=__UpperCAmelCase		).manual_seed(__UpperCAmelCase		)
         UpperCAmelCase__								=  {
             "image_embeds": image_embeds,
             "negative_image_embeds": negative_image_embeds,
             "hint": hint,
             "generator": generator,
             "height": 6_4,
             "width": 6_4,
             "guidance_scale": 4.0,
             "num_inference_steps": 2,
             "output_type": "np",
         }
         return inputs





   def       lowercase_				(self					:						Optional[Any]		)						->    str:
         """simple docstring"""



         UpperCAmelCase__								=  "cpu"

         UpperCAmelCase__								=  self.get_dummy_components()

         UpperCAmelCase__								=  self.pipeline_class(**__UpperCAmelCase		)
         UpperCAmelCase__								=  pipe.to(__UpperCAmelCase		)

         pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

         UpperCAmelCase__								=  pipe(**self.get_dummy_inputs(__UpperCAmelCase		)		)
         UpperCAmelCase__								=  output.images

         UpperCAmelCase__								=  pipe(
             **self.get_dummy_inputs(__UpperCAmelCase		)	,  return_dict=__UpperCAmelCase	,  )[0]

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

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

         UpperCAmelCase__								=  np.array(
             [0.6959826, 0.868279, 0.7558092, 0.68769467, 0.85805804, 0.65977496, 0.44885302, 0.5959111, 0.4251595]		)

         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 					A							(    unittest.TestCase					):


   def       lowercase_				(self					:						int		)						->    Optional[Any]:
         """simple docstring"""



         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





   def       lowercase_				(self					:						Tuple		)						->    int:
         """simple docstring"""



         UpperCAmelCase__								=  load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy"		)

         UpperCAmelCase__								=  load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/hint_image_cat.png"		)
         UpperCAmelCase__								=  torch.from_numpy(np.array(__UpperCAmelCase		)		).float() / 255.0
         UpperCAmelCase__								=  hint.permute(2	,  0	,  1		).unsqueeze(0		)

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

         UpperCAmelCase__								=  KandinskyVaaControlnetPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-controlnet-depth"	,  torch_dtype=torch.floataa		)
         UpperCAmelCase__								=  pipeline.to(__UpperCAmelCase		)
         pipeline.set_progress_bar_config(disable=__UpperCAmelCase		)

         UpperCAmelCase__								=  "A robot, 4k photo"

         UpperCAmelCase__								=  torch.Generator(device="cuda"		).manual_seed(0		)
         UpperCAmelCase__		,				UpperCAmelCase__								=  pipe_prior(
             __UpperCAmelCase	,  generator=__UpperCAmelCase	,  num_inference_steps=5	,  negative_prompt=""	,  ).to_tuple()

         UpperCAmelCase__								=  torch.Generator(device="cuda"		).manual_seed(0		)
         UpperCAmelCase__								=  pipeline(
             image_embeds=__UpperCAmelCase	,  negative_image_embeds=__UpperCAmelCase	,  hint=__UpperCAmelCase	,  generator=__UpperCAmelCase	,  num_inference_steps=1_0_0	,  output_type="np"	,  )

         UpperCAmelCase__								=  output.images[0]

         assert image.shape == (5_1_2, 5_1_2, 3)

         assert_mean_pixel_difference(__UpperCAmelCase	,  __UpperCAmelCase		)




from collections import Counter
from pathlib import Path
from typing import Optional, Tuple

import yaml





class 					A							(    yaml.SafeLoader					):


   def       lowercase_				(self					:						Tuple	,  __UpperCAmelCase					:						str		)						->    Tuple:
         """simple docstring"""



         UpperCAmelCase__								=  [self.constructed_objects[key_node] for key_node, _ in node.value]
         UpperCAmelCase__								=  [tuple(__UpperCAmelCase		) if isinstance(__UpperCAmelCase	,  __UpperCAmelCase		) else key for key in keys]
         UpperCAmelCase__								=  Counter(__UpperCAmelCase		)
         UpperCAmelCase__								=  [key for key in counter if counter[key] > 1]
         if duplicate_keys:
               raise TypeError(f"""Got duplicate yaml keys: {duplicate_keys}"""		)





   def       lowercase_				(self					:						Dict	,  __UpperCAmelCase					:						Dict	,  __UpperCAmelCase					:						Any=False		)						->    Union[str, Any]:
         """simple docstring"""



         UpperCAmelCase__								=  super().construct_mapping(__UpperCAmelCase	,  deep=__UpperCAmelCase		)
         self._check_no_duplicates_on_constructed_node(__UpperCAmelCase		)
         return mapping




def        lowerCAmelCase_			(					__A							)					->    Tuple[Optional[str], str]:







      '''simple docstring'''


      UpperCAmelCase__								=  list(readme_content.splitlines()							)
      if full_content and full_content[0] == "---" and "---" in full_content[1:]:
            UpperCAmelCase__								=  full_content[1:].index("---"							) + 1
            UpperCAmelCase__								=  "\n".join(full_content[1:sep_idx]							)
            return yamlblock, "\n".join(full_content[sep_idx + 1 :]							)

      return None, "\n".join(__A							)





class 					A							(    UpperCAmelCase_					):
   # class attributes
   __UpperCAmelCase			:			Optional[Any]              =      {'train_eval_index'}  # train-eval-index in the YAML metadata


   @classmethod
   def       lowercase_				(cls					:						List[str]	,  __UpperCAmelCase					:						Path		)						->    "DatasetMetadata":
         """simple docstring"""



         with open(__UpperCAmelCase	,  encoding="utf-8"		) as readme_file:
               UpperCAmelCase__		,				UpperCAmelCase__								=  _split_yaml_from_readme(readme_file.read()		)
         if yaml_string is not None:
               return cls.from_yaml_string(__UpperCAmelCase		)
         else:
               return cls()


   def       lowercase_				(self					:						int	,  __UpperCAmelCase					:						Path		)						->    str:
         """simple docstring"""



         if path.exists():
               with open(__UpperCAmelCase	,  encoding="utf-8"		) as readme_file:
                     UpperCAmelCase__								=  readme_file.read()
         else:
               UpperCAmelCase__								=  None
         UpperCAmelCase__								=  self._to_readme(__UpperCAmelCase		)
         with open(__UpperCAmelCase	,  "w"	,  encoding="utf-8"		) as readme_file:
               readme_file.write(__UpperCAmelCase		)


   def       lowercase_				(self					:						str	,  __UpperCAmelCase					:						Optional[str] = None		)						->    str:
         """simple docstring"""



         if readme_content is not None:
               UpperCAmelCase__		,				UpperCAmelCase__								=  _split_yaml_from_readme(__UpperCAmelCase		)
               UpperCAmelCase__								=  "---\n" + self.to_yaml_string() + "---\n" + content
         else:
               UpperCAmelCase__								=  "---\n" + self.to_yaml_string() + "---\n"
         return full_content


   @classmethod
   def       lowercase_				(cls					:						Optional[int]	,  __UpperCAmelCase					:						str		)						->    "DatasetMetadata":
         """simple docstring"""



         UpperCAmelCase__								=  yaml.load(__UpperCAmelCase	,  Loader=_NoDuplicateSafeLoader		) or {}

         # Convert the YAML keys to DatasetMetadata fields
         UpperCAmelCase__								=  {
             (key.replace("-"	,  "_"		) if key.replace("-"	,  "_"		) in cls._FIELDS_WITH_DASHES else key): value
             for key, value in metadata_dict.items()
         }
         return cls(**__UpperCAmelCase		)





   def       lowercase_				(self					:						List[str]		)						->    str:
         """simple docstring"""



         return yaml.safe_dump(
             {
                 (key.replace("_"	,  "-"		) if key in self._FIELDS_WITH_DASHES else key): value
                 for key, value in self.items()
             }	,  sort_keys=__UpperCAmelCase	,  allow_unicode=__UpperCAmelCase	,  encoding="utf-8"	,  ).decode("utf-8"		)




UpperCamelCase__    							= {
    'image-classification': [],
    'translation': [],
    'image-segmentation': [],
    'fill-mask': [],
    'automatic-speech-recognition': [],
    'token-classification': [],
    'sentence-similarity': [],
    'audio-classification': [],
    'question-answering': [],
    'summarization': [],
    'zero-shot-classification': [],
    'table-to-text': [],
    'feature-extraction': [],
    'other': [],
    'multiple-choice': [],
    'text-classification': [],
    'text-to-image': [],
    'text2text-generation': [],
    'zero-shot-image-classification': [],
    'tabular-classification': [],
    'tabular-regression': [],
    'image-to-image': [],
    'tabular-to-text': [],
    'unconditional-image-generation': [],
    'text-retrieval': [],
    'text-to-speech': [],
    'object-detection': [],
    'audio-to-audio': [],
    'text-generation': [],
    'conversational': [],
    'table-question-answering': [],
    'visual-question-answering': [],
    'image-to-text': [],
    'reinforcement-learning': [],
    'voice-activity-detection': [],
    'time-series-forecasting': [],
    'document-question-answering': [],
}


if __name__ == "__main__":
      from argparse import ArgumentParser

      UpperCamelCase__    							= ArgumentParser(usage='Validate the yaml metadata block of a README.md file.')
      ap.add_argument('readme_filepath')
      UpperCamelCase__    							= ap.parse_args()

      UpperCamelCase__    							= Path(args.readme_filepath)
      UpperCamelCase__    							= DatasetMetadata.from_readme(readme_filepath)
      print(dataset_metadata)
      dataset_metadata.to_readme(readme_filepath)



