





from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin

@dataclass
class      A       (	lowercase_						):
      _snake_case							     =42

class      A       (	lowercase_		,      lowercase_						):



      @register_to_config
      def __init__(	self:						Optional[int]		,  _lowerCAmelCase:						int = 32		,  _lowerCAmelCase:						int = 64		,  _lowerCAmelCase:						int = 20		,  _lowerCAmelCase:						int = 768		,  _lowerCAmelCase:						Union[str, Any]=77		,  _lowerCAmelCase:						List[Any]=4		,  _lowerCAmelCase:						float = 0.0		,  _lowerCAmelCase:						str = "silu"		,  _lowerCAmelCase:						Optional[str] = None		,  _lowerCAmelCase:						Optional[str] = None		,  _lowerCAmelCase:						Optional[str] = "linear"		,  _lowerCAmelCase:						Optional[str] = "prd"		,  _lowerCAmelCase:						Optional[int] = None		,  _lowerCAmelCase:						Optional[int] = None		,  _lowerCAmelCase:						Optional[int] = None		,  )						->    int:
         '''simple docstring'''

         super().__init__()
         UpperCAmelCase_											=num_attention_heads
         UpperCAmelCase_											=attention_head_dim
         UpperCAmelCase_											=num_attention_heads * attention_head_dim
         UpperCAmelCase_											=additional_embeddings

         UpperCAmelCase_											=time_embed_dim or inner_dim
         UpperCAmelCase_											=embedding_proj_dim or embedding_dim
         UpperCAmelCase_											=clip_embed_dim or embedding_dim

         UpperCAmelCase_											=Timesteps(lowerCamelCase_		,  lowerCamelCase_		,  0       )
         UpperCAmelCase_											=TimestepEmbedding(lowerCamelCase_		,  lowerCamelCase_		,  out_dim=lowerCamelCase_		,  act_fn=lowerCamelCase_       )

         UpperCAmelCase_											=nn.Linear(lowerCamelCase_		,  lowerCamelCase_       )

         if embedding_proj_norm_type is None:
            UpperCAmelCase_											=None
         elif embedding_proj_norm_type == "layer":
            UpperCAmelCase_											=nn.LayerNorm(lowerCamelCase_       )
         else:
            raise ValueError(F'unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'       )

         UpperCAmelCase_											=nn.Linear(lowerCamelCase_		,  lowerCamelCase_       )

         if encoder_hid_proj_type is None:
            UpperCAmelCase_											=None
         elif encoder_hid_proj_type == "linear":
            UpperCAmelCase_											=nn.Linear(lowerCamelCase_		,  lowerCamelCase_       )
         else:
            raise ValueError(F'unsupported encoder_hid_proj_type: {encoder_hid_proj_type}'       )

         UpperCAmelCase_											=nn.Parameter(torch.zeros(1		,  num_embeddings + additional_embeddings		,  lowerCamelCase_       )       )

         if added_emb_type == "prd":
            UpperCAmelCase_											=nn.Parameter(torch.zeros(1		,  1		,  lowerCamelCase_       )       )
         elif added_emb_type is None:
            UpperCAmelCase_											=None
         else:
            raise ValueError(
                F'`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'       )

         UpperCAmelCase_											=nn.ModuleList(
             [
                 BasicTransformerBlock(
                     lowerCamelCase_		,  lowerCamelCase_		,  lowerCamelCase_		,  dropout=lowerCamelCase_		,  activation_fn="gelu"		,  attention_bias=lowerCamelCase_		,  )
                 for d in range(lowerCamelCase_       )
             ]       )

         if norm_in_type == "layer":
            UpperCAmelCase_											=nn.LayerNorm(lowerCamelCase_       )
         elif norm_in_type is None:
            UpperCAmelCase_											=None
         else:
            raise ValueError(F'Unsupported norm_in_type: {norm_in_type}.'       )

         UpperCAmelCase_											=nn.LayerNorm(lowerCamelCase_       )

         UpperCAmelCase_											=nn.Linear(lowerCamelCase_		,  lowerCamelCase_       )

         UpperCAmelCase_											=torch.full(
             [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]		,  -1_0000.0       )
         causal_attention_mask.triu_(1       )
         UpperCAmelCase_											=causal_attention_mask[None, ...]
         self.register_buffer("causal_attention_mask"		,  lowerCamelCase_		,  persistent=lowerCamelCase_       )

         UpperCAmelCase_											=nn.Parameter(torch.zeros(1		,  lowerCamelCase_       )       )
         UpperCAmelCase_											=nn.Parameter(torch.zeros(1		,  lowerCamelCase_       )       )



      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def 	lowerCAmelCase__    (	self:						List[Any]       )						->    Tuple:
         '''simple docstring'''

         UpperCAmelCase_											={}

         def fn_recursive_add_processors(_lowerCAmelCase:						str		,  _lowerCAmelCase:						torch.nn.Module		,  _lowerCAmelCase:						Dict[str, AttentionProcessor]       ):
            if hasattr(lowerCamelCase_		,  "set_processor"       ):
               UpperCAmelCase_											=module.processor

            for sub_name, child in module.named_children():
               fn_recursive_add_processors(F'{name}.{sub_name}'		,  lowerCamelCase_		,  lowerCamelCase_       )

            return processors

         for name, module in self.named_children():
            fn_recursive_add_processors(lowerCamelCase_		,  lowerCamelCase_		,  lowerCamelCase_       )

         return processors



      def 	lowerCAmelCase__    (	self:						List[str]		,  _lowerCAmelCase:						Union[AttentionProcessor, Dict[str, AttentionProcessor]]       )						->    str:
         '''simple docstring'''

         UpperCAmelCase_											=len(self.attn_processors.keys()       )

         if isinstance(lowerCamelCase_		,  lowerCamelCase_       ) and len(lowerCamelCase_       ) != count:
            raise ValueError(
                F'A dict of processors was passed, but the number of processors {len(lowerCamelCase_       )} does not match the'
                F' number of attention layers: {count}. Please make sure to pass {count} processor classes.'       )

         def fn_recursive_attn_processor(_lowerCAmelCase:						str		,  _lowerCAmelCase:						torch.nn.Module		,  _lowerCAmelCase:						Optional[Any]       ):
            if hasattr(lowerCamelCase_		,  "set_processor"       ):
               if not isinstance(lowerCamelCase_		,  lowerCamelCase_       ):
                  module.set_processor(lowerCamelCase_       )
               else:
                  module.set_processor(processor.pop(F'{name}.processor'       )       )

            for sub_name, child in module.named_children():
               fn_recursive_attn_processor(F'{name}.{sub_name}'		,  lowerCamelCase_		,  lowerCamelCase_       )

         for name, module in self.named_children():
            fn_recursive_attn_processor(lowerCamelCase_		,  lowerCamelCase_		,  lowerCamelCase_       )



      def 	lowerCAmelCase__    (	self:						Any       )						->    Any:
         '''simple docstring'''

         self.set_attn_processor(AttnProcessor()       )



      def 	lowerCAmelCase__    (	self:						Optional[int]		,  _lowerCAmelCase:						Tuple		,  _lowerCAmelCase:						Union[torch.Tensor, float, int]		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						Optional[torch.FloatTensor] = None		,  _lowerCAmelCase:						Optional[torch.BoolTensor] = None		,  _lowerCAmelCase:						bool = True		,  )						->    List[str]:
         '''simple docstring'''

         UpperCAmelCase_											=hidden_states.shape[0]

         UpperCAmelCase_											=timestep
         if not torch.is_tensor(lowerCamelCase_       ):
            UpperCAmelCase_											=torch.tensor([timesteps]		,  dtype=torch.long		,  device=hidden_states.device       )
         elif torch.is_tensor(lowerCamelCase_       ) and len(timesteps.shape       ) == 0:
            UpperCAmelCase_											=timesteps[None].to(hidden_states.device       )

         # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
         UpperCAmelCase_											=timesteps * torch.ones(lowerCamelCase_		,  dtype=timesteps.dtype		,  device=timesteps.device       )

         UpperCAmelCase_											=self.time_proj(lowerCamelCase_       )

         # timesteps does not contain any weights and will always return f32 tensors
         # but time_embedding might be fp16, so we need to cast here.
         UpperCAmelCase_											=timesteps_projected.to(dtype=self.dtype       )
         UpperCAmelCase_											=self.time_embedding(lowerCamelCase_       )

         if self.embedding_proj_norm is not None:
            UpperCAmelCase_											=self.embedding_proj_norm(lowerCamelCase_       )

         UpperCAmelCase_											=self.embedding_proj(lowerCamelCase_       )
         if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
            UpperCAmelCase_											=self.encoder_hidden_states_proj(lowerCamelCase_       )
         elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
            raise ValueError("`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set"       )

         UpperCAmelCase_											=self.proj_in(lowerCamelCase_       )

         UpperCAmelCase_											=self.positional_embedding.to(hidden_states.dtype       )

         UpperCAmelCase_											=[]
         UpperCAmelCase_											=0

         if encoder_hidden_states is not None:
            additional_embeds.append(lowerCamelCase_       )
            additional_embeddings_len += encoder_hidden_states.shape[1]

         if len(proj_embeddings.shape       ) == 2:
            UpperCAmelCase_											=proj_embeddings[:, None, :]

         if len(hidden_states.shape       ) == 2:
            UpperCAmelCase_											=hidden_states[:, None, :]

         UpperCAmelCase_											=additional_embeds + [
             proj_embeddings,
             time_embeddings[:, None, :],
             hidden_states,
         ]

         if self.prd_embedding is not None:
            UpperCAmelCase_											=self.prd_embedding.to(hidden_states.dtype       ).expand(lowerCamelCase_		,  -1		,  -1       )
            additional_embeds.append(lowerCamelCase_       )

         UpperCAmelCase_											=torch.cat(
             lowerCamelCase_		,  dim=1		,  )

         # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
         UpperCAmelCase_											=additional_embeddings_len + proj_embeddings.shape[1] + 1
         if positional_embeddings.shape[1] < hidden_states.shape[1]:
            UpperCAmelCase_											=F.pad(
                lowerCamelCase_		,  (
                    0,
                    0,
                    additional_embeddings_len,
                    self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
                )		,  value=0.0		,  )

         UpperCAmelCase_											=hidden_states + positional_embeddings

         if attention_mask is not None:
            UpperCAmelCase_											=(1 - attention_mask.to(hidden_states.dtype       )) * -1_0000.0
            UpperCAmelCase_											=F.pad(lowerCamelCase_		,  (0, self.additional_embeddings)		,  value=0.0       )
            UpperCAmelCase_											=(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype       )
            UpperCAmelCase_											=attention_mask.repeat_interleave(self.config.num_attention_heads		,  dim=0       )

         if self.norm_in is not None:
            UpperCAmelCase_											=self.norm_in(lowerCamelCase_       )

         for block in self.transformer_blocks:
            UpperCAmelCase_											=block(lowerCamelCase_		,  attention_mask=lowerCamelCase_       )

         UpperCAmelCase_											=self.norm_out(lowerCamelCase_       )

         if self.prd_embedding is not None:
            UpperCAmelCase_											=hidden_states[:, -1]
         else:
            UpperCAmelCase_											=hidden_states[:, additional_embeddings_len:]

         UpperCAmelCase_											=self.proj_to_clip_embeddings(lowerCamelCase_       )

         if not return_dict:
            return (predicted_image_embedding,)

         return PriorTransformerOutput(predicted_image_embedding=lowerCamelCase_       )






      def 	lowerCAmelCase__    (	self:						Dict		,  _lowerCAmelCase:						Tuple       )						->    Optional[int]:
         '''simple docstring'''

         UpperCAmelCase_											=(prior_latents * self.clip_std) + self.clip_mean
         return prior_latents











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

import yaml

class      A       (	yaml.SafeLoader						):



      def 	lowerCAmelCase__    (	self:						Optional[Any]		,  _lowerCAmelCase:						int       )						->    Optional[int]:
         '''simple docstring'''

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






      def 	lowerCAmelCase__    (	self:						Dict		,  _lowerCAmelCase:						List[str]		,  _lowerCAmelCase:						Optional[Any]=False       )						->    int:
         '''simple docstring'''

         UpperCAmelCase_											=super().construct_mapping(_lowerCAmelCase		,  deep=_lowerCAmelCase       )
         self._check_no_duplicates_on_constructed_node(_lowerCAmelCase       )
         return mapping




def     a__ (					lowercase__  ):






   '''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(lowercase__  )

class      A       (	__lowercase						):
      # class attributes
      _snake_case							     ={'''train_eval_index'''}  # train-eval-index in the YAML metadata



      @classmethod
      def 	lowerCAmelCase__    (	cls:						Any		,  _lowerCAmelCase:						Path       )						->    "DatasetMetadata":
         '''simple docstring'''

         with open(_lowerCAmelCase		,  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(_lowerCAmelCase       )
         else:
            return cls()



      def 	lowerCAmelCase__    (	self:						Dict		,  _lowerCAmelCase:						Path       )						->    List[str]:
         '''simple docstring'''

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



      def 	lowerCAmelCase__    (	self:						str		,  _lowerCAmelCase:						Optional[str] = None       )						->    str:
         '''simple docstring'''

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



      @classmethod
      def 	lowerCAmelCase__    (	cls:						Optional[Any]		,  _lowerCAmelCase:						str       )						->    "DatasetMetadata":
         '''simple docstring'''

         UpperCAmelCase_											=yaml.load(_lowerCAmelCase		,  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(**_lowerCAmelCase       )






      def 	lowerCAmelCase__    (	self:						int       )						->    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=_lowerCAmelCase		,  allow_unicode=_lowerCAmelCase		,  encoding="utf-8"		,  ).decode("utf-8"       )

__lowercase  :   int						      ={
    """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

     __lowercase  :   List[Any]						      =ArgumentParser(usage="""Validate the yaml metadata block of a README.md file.""")
     ap.add_argument("""readme_filepath""")
     __lowercase  :   int						      =ap.parse_args()

     __lowercase  :   Optional[int]						      =Path(args.readme_filepath)
     __lowercase  :   Tuple						      =DatasetMetadata.from_readme(readme_filepath)
     print(dataset_metadata)
     dataset_metadata.to_readme(readme_filepath)




