




'''simple docstring'''





from typing import Any, Dict, Optional

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

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings

@maybe_allow_in_graph
class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=0.0 ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = "geglu" ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = False ,_SCREAMING_SNAKE_CASE = False ,_SCREAMING_SNAKE_CASE = False ,_SCREAMING_SNAKE_CASE = False ,_SCREAMING_SNAKE_CASE = True ,_SCREAMING_SNAKE_CASE = "layer_norm" ,_SCREAMING_SNAKE_CASE = False ,)							->      str:
          super().__init__()
          _snake_case							=				only_cross_attention

          _snake_case							=				(num_embeds_ada_norm is not None) and norm_type == "ada_norm_zero"
          _snake_case							=				(num_embeds_ada_norm is not None) and norm_type == "ada_norm"

          if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
               raise ValueError(
                   f"""`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to"""
                   f""" define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}."""   )

          # Define 3 blocks. Each block has its own normalization layer.
          # 1. Self-Attn
          if self.use_ada_layer_norm:
               _snake_case							=				AdaLayerNorm(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          elif self.use_ada_layer_norm_zero:
               _snake_case							=				AdaLayerNormZero(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          else:
               _snake_case							=				nn.LayerNorm(_SCREAMING_SNAKE_CASE ,elementwise_affine=_SCREAMING_SNAKE_CASE   )
          _snake_case							=				Attention(
              query_dim=_SCREAMING_SNAKE_CASE ,heads=_SCREAMING_SNAKE_CASE ,dim_head=_SCREAMING_SNAKE_CASE ,dropout=_SCREAMING_SNAKE_CASE ,bias=_SCREAMING_SNAKE_CASE ,cross_attention_dim=cross_attention_dim if only_cross_attention else None ,upcast_attention=_SCREAMING_SNAKE_CASE ,)

          # 2. Cross-Attn
          if cross_attention_dim is not None or double_self_attention:
               # We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
               # I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
               # the second cross attention block.
               _snake_case							=				(
                   AdaLayerNorm(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
                   if self.use_ada_layer_norm
                   else nn.LayerNorm(_SCREAMING_SNAKE_CASE ,elementwise_affine=_SCREAMING_SNAKE_CASE   )
               )
               _snake_case							=				Attention(
                   query_dim=_SCREAMING_SNAKE_CASE ,cross_attention_dim=cross_attention_dim if not double_self_attention else None ,heads=_SCREAMING_SNAKE_CASE ,dim_head=_SCREAMING_SNAKE_CASE ,dropout=_SCREAMING_SNAKE_CASE ,bias=_SCREAMING_SNAKE_CASE ,upcast_attention=_SCREAMING_SNAKE_CASE ,)  # is self-attn if encoder_hidden_states is none
          else:
               _snake_case							=				None
               _snake_case							=				None

          # 3. Feed-forward
          _snake_case							=				nn.LayerNorm(_SCREAMING_SNAKE_CASE ,elementwise_affine=_SCREAMING_SNAKE_CASE   )
          _snake_case							=				FeedForward(_SCREAMING_SNAKE_CASE ,dropout=_SCREAMING_SNAKE_CASE ,activation_fn=_SCREAMING_SNAKE_CASE ,final_dropout=_SCREAMING_SNAKE_CASE   )

          # let chunk size default to None
          _snake_case							=				None
          _snake_case							=				0






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Dict:
          # Sets chunk feed-forward
          _snake_case							=				chunk_size
          _snake_case							=				dim






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = None ,)							->      Dict:
          # Notice that normalization is always applied before the real computation in the following blocks.
          # 1. Self-Attention
          if self.use_ada_layer_norm:
               _snake_case							=				self.norma(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          elif self.use_ada_layer_norm_zero:
               _snake_case							, _snake_case							, _snake_case							, _snake_case							, _snake_case							=				self.norma(
                   _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,hidden_dtype=hidden_states.dtype   )
          else:
               _snake_case							=				self.norma(_SCREAMING_SNAKE_CASE   )

          _snake_case							=				cross_attention_kwargs if cross_attention_kwargs is not None else {}

          _snake_case							=				self.attna(
              _SCREAMING_SNAKE_CASE ,encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None ,attention_mask=_SCREAMING_SNAKE_CASE ,**_SCREAMING_SNAKE_CASE ,)
          if self.use_ada_layer_norm_zero:
               _snake_case							=				gate_msa.unsqueeze(1   ) * attn_output
          _snake_case							=				attn_output + hidden_states

          # 2. Cross-Attention
          if self.attna is not None:
               _snake_case							=				(
                   self.norma(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   ) if self.use_ada_layer_norm else self.norma(_SCREAMING_SNAKE_CASE   )
               )

               _snake_case							=				self.attna(
                   _SCREAMING_SNAKE_CASE ,encoder_hidden_states=_SCREAMING_SNAKE_CASE ,attention_mask=_SCREAMING_SNAKE_CASE ,**_SCREAMING_SNAKE_CASE ,)
               _snake_case							=				attn_output + hidden_states

          # 3. Feed-forward
          _snake_case							=				self.norma(_SCREAMING_SNAKE_CASE   )

          if self.use_ada_layer_norm_zero:
               _snake_case							=				norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

          if self._chunk_size is not None:
               # "feed_forward_chunk_size" can be used to save memory
               if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
                    raise ValueError(
                        f"""`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`."""   )

               _snake_case							=				norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
               _snake_case							=				torch.cat(
                   [self.ff(_SCREAMING_SNAKE_CASE   ) for hid_slice in norm_hidden_states.chunk(_SCREAMING_SNAKE_CASE ,dim=self._chunk_dim   )] ,dim=self._chunk_dim ,)
          else:
               _snake_case							=				self.ff(_SCREAMING_SNAKE_CASE   )

          if self.use_ada_layer_norm_zero:
               _snake_case							=				gate_mlp.unsqueeze(1   ) * ff_output

          _snake_case							=				ff_output + hidden_states

          return hidden_states







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = 4 ,_SCREAMING_SNAKE_CASE = 0.0 ,_SCREAMING_SNAKE_CASE = "geglu" ,_SCREAMING_SNAKE_CASE = False ,)							->      Union[str, Any]:
          super().__init__()
          _snake_case							=				int(dim * mult   )
          _snake_case							=				dim_out if dim_out is not None else dim

          if activation_fn == "gelu":
               _snake_case							=				GELU(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          if activation_fn == "gelu-approximate":
               _snake_case							=				GELU(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,approximate="tanh"   )
          elif activation_fn == "geglu":
               _snake_case							=				GEGLU(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          elif activation_fn == "geglu-approximate":
               _snake_case							=				ApproximateGELU(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )

          _snake_case							=				nn.ModuleList([]   )
          # project in
          self.net.append(_SCREAMING_SNAKE_CASE   )
          # project dropout
          self.net.append(nn.Dropout(_SCREAMING_SNAKE_CASE   )   )
          # project out
          self.net.append(nn.Linear(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )   )
          # FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
          if final_dropout:
               self.net.append(nn.Dropout(_SCREAMING_SNAKE_CASE   )   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      int:
          for module in self.net:
               _snake_case							=				module(_SCREAMING_SNAKE_CASE   )
          return hidden_states







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE = "none"   )							->      Dict:
          super().__init__()
          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							=				approximate






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      List[str]:
          if gate.device.type != "mps":
               return F.gelu(_SCREAMING_SNAKE_CASE ,approximate=self.approximate   )
          # mps: gelu is not implemented for float16
          return F.gelu(gate.to(dtype=torch.floataa   ) ,approximate=self.approximate   ).to(dtype=gate.dtype   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      int:
          _snake_case							=				self.proj(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				self.gelu(_SCREAMING_SNAKE_CASE   )
          return hidden_states







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Optional[Any]:
          super().__init__()
          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,dim_out * 2   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      Optional[int]:
          if gate.device.type != "mps":
               return F.gelu(_SCREAMING_SNAKE_CASE   )
          # mps: gelu is not implemented for float16
          return F.gelu(gate.to(dtype=torch.floataa   )   ).to(dtype=gate.dtype   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      int:
          _snake_case							, _snake_case							=				self.proj(_SCREAMING_SNAKE_CASE   ).chunk(2 ,dim=-1   )
          return hidden_states * self.gelu(_SCREAMING_SNAKE_CASE   )







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      List[Any]:
          super().__init__()
          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      Optional[Any]:
          _snake_case							=				self.proj(_SCREAMING_SNAKE_CASE   )
          return x * torch.sigmoid(1.7_0_2 * x   )







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      str:
          super().__init__()
          _snake_case							=				nn.Embedding(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							=				nn.SiLU()
          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,embedding_dim * 2   )
          _snake_case							=				nn.LayerNorm(_SCREAMING_SNAKE_CASE ,elementwise_affine=_SCREAMING_SNAKE_CASE   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      int:
          _snake_case							=				self.linear(self.silu(self.emb(_SCREAMING_SNAKE_CASE   )   )   )
          _snake_case							, _snake_case							=				torch.chunk(_SCREAMING_SNAKE_CASE ,2   )
          _snake_case							=				self.norm(_SCREAMING_SNAKE_CASE   ) * (1 + scale) + shift
          return x







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Union[str, Any]:
          super().__init__()

          _snake_case							=				CombinedTimestepLabelEmbeddings(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )

          _snake_case							=				nn.SiLU()
          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,6 * embedding_dim ,bias=_SCREAMING_SNAKE_CASE   )
          _snake_case							=				nn.LayerNorm(_SCREAMING_SNAKE_CASE ,elementwise_affine=_SCREAMING_SNAKE_CASE ,eps=1e-6   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=None   )							->      str:
          _snake_case							=				self.linear(self.silu(self.emb(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,hidden_dtype=_SCREAMING_SNAKE_CASE   )   )   )
          _snake_case							, _snake_case							, _snake_case							, _snake_case							, _snake_case							, _snake_case							=				emb.chunk(6 ,dim=1   )
          _snake_case							=				self.norm(_SCREAMING_SNAKE_CASE   ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
          return x, gate_msa, shift_mlp, scale_mlp, gate_mlp







class        _a    (						nn.Module							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE = None ,_SCREAMING_SNAKE_CASE = 1e-5   )							->      str:
          super().__init__()
          _snake_case							=				num_groups
          _snake_case							=				eps

          if act_fn is None:
               _snake_case							=				None
          else:
               _snake_case							=				get_activation(_SCREAMING_SNAKE_CASE   )

          _snake_case							=				nn.Linear(_SCREAMING_SNAKE_CASE ,out_dim * 2   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Union[str, Any]:
          if self.act:
               _snake_case							=				self.act(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				self.linear(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				emb[:, :, None, None]
          _snake_case							, _snake_case							=				emb.chunk(2 ,dim=1   )

          _snake_case							=				F.group_norm(_SCREAMING_SNAKE_CASE ,self.num_groups ,eps=self.eps   )
          _snake_case							=				x * (1 + scale) + shift
          return x








'''simple docstring'''





import argparse
import gc
import json
import os
import re

import torch
from huggingface_hub import hf_hub_download

from transformers import AutoModelForCausalLM, AutoTokenizer, PreTrainedTokenizerFast, RwkvConfig
from transformers.modeling_utils import WEIGHTS_INDEX_NAME, shard_checkpoint


UpperCamelCase_						:      int       =						{
    '''169M''': 12,
    '''430M''': 24,
    '''1B5''': 24,
    '''3B''': 32,
    '''7B''': 32,
    '''14B''': 40,
}

UpperCamelCase_						:      str       =						{
    '''169M''': 768,
    '''430M''': 1024,
    '''1B5''': 2048,
    '''3B''': 2560,
    '''7B''': 4096,
    '''14B''': 5120,
}





def      __a     (       _UpperCamelCase:   str   )						->			Any:

     """simple docstring"""




     _snake_case							=				list(state_dict.keys()   )
     for name in state_dict_keys:
          _snake_case							=				state_dict.pop(_UpperCamelCase   )
          # emb -> embedding
          if name.startswith("emb."   ):
               _snake_case							=				name.replace("emb." ,						"embeddings."   )
          # ln_0 -> pre_ln (only present at block 0)
          if name.startswith("blocks.0.ln0"   ):
               _snake_case							=				name.replace("blocks.0.ln0" ,						"blocks.0.pre_ln"   )
          # att -> attention
          _snake_case							=				re.sub(r"blocks\.(\d+)\.att" ,						r"blocks.\1.attention" ,						_UpperCamelCase   )
          # ffn -> feed_forward
          _snake_case							=				re.sub(r"blocks\.(\d+)\.ffn" ,						r"blocks.\1.feed_forward" ,						_UpperCamelCase   )
          # time_mix_k -> time_mix_key and reshape
          if name.endswith(".time_mix_k"   ):
               _snake_case							=				name.replace(".time_mix_k" ,						".time_mix_key"   )
          # time_mix_v -> time_mix_value and reshape
          if name.endswith(".time_mix_v"   ):
               _snake_case							=				name.replace(".time_mix_v" ,						".time_mix_value"   )
          # time_mix_r -> time_mix_key and reshape
          if name.endswith(".time_mix_r"   ):
               _snake_case							=				name.replace(".time_mix_r" ,						".time_mix_receptance"   )

          if name != "head.weight":
               _snake_case							=				"rwkv." + name

          _snake_case							=				weight
     return state_dict





def      __a     (       _UpperCamelCase:   Any ,						_UpperCamelCase:   List[Any] ,						_UpperCamelCase:   List[Any] ,						_UpperCamelCase:   str=None ,						_UpperCamelCase:   Optional[Any]=None ,						_UpperCamelCase:   List[str]=False ,						_UpperCamelCase:   Dict=None   )						->			Dict:

     """simple docstring"""




     if tokenizer_file is None:
          print("No `--tokenizer_file` provided, we will use the default tokenizer."   )
          _snake_case							=				50_277
          _snake_case							=				AutoTokenizer.from_pretrained("EleutherAI/gpt-neox-20b"   )
     else:
          _snake_case							=				PreTrainedTokenizerFast(tokenizer_file=_UpperCamelCase   )
          _snake_case							=				len(_UpperCamelCase   )
     tokenizer.save_pretrained(_UpperCamelCase   )

     # 2. Build the config
     _snake_case							=				list(NUM_HIDDEN_LAYERS_MAPPING.keys()   )
     if size is None:
          # Try to infer size from the checkpoint name
          for candidate in possible_sizes:
               if candidate in checkpoint_file:
                    _snake_case							=				candidate
                    break
          if size is None:
               raise ValueError("Could not infer the size, please provide it with the `--size` argument."   )
     if size not in possible_sizes:
          raise ValueError(F"""`size` should be one of {possible_sizes}, got {size}."""   )

     _snake_case							=				RwkvConfig(
         vocab_size=_UpperCamelCase ,						num_hidden_layers=NUM_HIDDEN_LAYERS_MAPPING[size] ,						hidden_size=HIDEN_SIZE_MAPPING[size] ,						)
     config.save_pretrained(_UpperCamelCase   )

     # 3. Download model file then convert state_dict
     _snake_case							=				hf_hub_download(_UpperCamelCase ,						_UpperCamelCase   )
     _snake_case							=				torch.load(_UpperCamelCase ,						map_location="cpu"   )
     _snake_case							=				convert_state_dict(_UpperCamelCase   )

     # 4. Split in shards and save
     _snake_case							, _snake_case							=				shard_checkpoint(_UpperCamelCase   )
     for shard_file, shard in shards.items():
          torch.save(_UpperCamelCase ,						os.path.join(_UpperCamelCase ,						_UpperCamelCase   )   )

     if index is not None:
          _snake_case							=				os.path.join(_UpperCamelCase ,						_UpperCamelCase   )
          # Save the index as well
          with open(_UpperCamelCase ,						"w" ,						encoding="utf-8"   ) as f:
               _snake_case							=				json.dumps(_UpperCamelCase ,						indent=2 ,						sort_keys=_UpperCamelCase   ) + "\n"
               f.write(_UpperCamelCase   )

          # 5. Clean up shards (for some reason the file PyTorch saves take the same space as the whole state_dict
          print(
              "Cleaning up shards. This may error with an OOM error, it this is the case don't worry you still have converted the model."   )
          _snake_case							=				list(shards.keys()   )

          del state_dict
          del shards
          gc.collect()

          for shard_file in shard_files:
               _snake_case							=				torch.load(os.path.join(_UpperCamelCase ,						_UpperCamelCase   )   )
               torch.save({k: v.cpu().clone() for k, v in state_dict.items()} ,						os.path.join(_UpperCamelCase ,						_UpperCamelCase   )   )

     del state_dict
     gc.collect()

     if push_to_hub:
          if model_name is None:
               raise ValueError("Please provide a `model_name` to push the model to the Hub."   )
          _snake_case							=				AutoModelForCausalLM.from_pretrained(_UpperCamelCase   )
          model.push_to_hub(_UpperCamelCase ,						max_shard_size="2GB"   )
          tokenizer.push_to_hub(_UpperCamelCase   )


if __name__ == "__main__":
   UpperCamelCase_						:      Any       =						argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       '''--repo_id''', default=None, type=str, required=True, help='''Repo ID from which to pull the checkpoint.'''
   )
   parser.add_argument(
       '''--checkpoint_file''', default=None, type=str, required=True, help='''Name of the checkpoint file in the repo.'''
   )
   parser.add_argument(
       '''--output_dir''', default=None, type=str, required=True, help='''Where to save the converted model.'''
   )
   parser.add_argument(
       '''--tokenizer_file''',
       default=None,
       type=str,
       help='''Path to the tokenizer file to use (if not provided, only the model is converted).''',
   )
   parser.add_argument(
       '''--size''',
       default=None,
       type=str,
       help='''Size of the model. Will be inferred from the `checkpoint_file` if not passed.''',
   )
   parser.add_argument(
       '''--push_to_hub''',
       action='''store_true''',
       help='''Push to the Hub the converted model.''',
   )
   parser.add_argument(
       '''--model_name''',
       default=None,
       type=str,
       help='''Name of the pushed model on the Hub, including the username / organization.''',
   )

   UpperCamelCase_						:      int       =						parser.parse_args()
   convert_rmkv_checkpoint_to_hf_format(
       args.repo_id,
       args.checkpoint_file,
       args.output_dir,
       size=args.size,
       tokenizer_file=args.tokenizer_file,
       push_to_hub=args.push_to_hub,
       model_name=args.model_name,
   )


