





import argparse
import re

import torch
from CLAP import create_model

from transformers import AutoFeatureExtractor, ClapConfig, ClapModel


snake_case_				=			{
    'text_branch': 'text_model',
    'audio_branch': 'audio_model.audio_encoder',
    'attn': 'attention.self',
    'self.proj': 'output.dense',
    'attention.self_mask': 'attn_mask',
    'mlp.fc1': 'intermediate.dense',
    'mlp.fc2': 'output.dense',
    'norm1': 'layernorm_before',
    'norm2': 'layernorm_after',
    'bn0': 'batch_norm',
}

snake_case_				=			AutoFeatureExtractor.from_pretrained('laion/clap-htsat-unfused', truncation='rand_trunc')






def lowerCamelCase__    (							snake_case_	:  str       ,      snake_case_	:  Any=False       )      ->				Optional[Any]:
   __snake_case       ,	__snake_case							       =   create_model(
       '''HTSAT-tiny'''       ,      '''roberta'''       ,      snake_case_       ,      precision='''fp32'''       ,      device='''cuda:0''' if torch.cuda.is_available() else '''cpu'''       ,      enable_fusion=snake_case_       ,      fusion_type='''aff_2d''' if enable_fusion else None       ,      )
   return model, model_cfg






def lowerCamelCase__    (							snake_case_	:  Dict       )      ->				Any:
   __snake_case							       =   {}

   __snake_case							       =   R'''.*sequential.(\d+).*'''
   __snake_case							       =   R'''.*_projection.(\d+).*'''

   for key, value in state_dict.items():
      # check if any key needs to be modified
      for key_to_modify, new_key in KEYS_TO_MODIFY_MAPPING.items():
         if key_to_modify in key:
            __snake_case							       =   key.replace(snake_case_       ,      snake_case_       )

      if re.match(snake_case_       ,      snake_case_       ):
         # replace sequential layers with list
         __snake_case							       =   re.match(snake_case_       ,      snake_case_       ).group(1       )

         __snake_case							       =   key.replace(f"""sequential.{sequential_layer}."""       ,      f"""layers.{int(snake_case_       )//3}.linear."""       )
      elif re.match(snake_case_       ,      snake_case_       ):
         __snake_case							       =   int(re.match(snake_case_       ,      snake_case_       ).group(1       )       )

         # Because in CLAP they use `nn.Sequential`...
         __snake_case							       =   1 if projecton_layer == 0 else 2

         __snake_case							       =   key.replace(f"""_projection.{projecton_layer}."""       ,      f"""_projection.linear{transformers_projection_layer}."""       )

      if "audio" and "qkv" in key:
         # split qkv into query key and value
         __snake_case							       =   value
         __snake_case							       =   mixed_qkv.size(0       ) // 3

         __snake_case							       =   mixed_qkv[:qkv_dim]
         __snake_case							       =   mixed_qkv[qkv_dim : qkv_dim * 2]
         __snake_case							       =   mixed_qkv[qkv_dim * 2 :]

         __snake_case							       =   query_layer
         __snake_case							       =   key_layer
         __snake_case							       =   value_layer
      else:
         __snake_case							       =   value

   return model_state_dict






def lowerCamelCase__    (							snake_case_	:  str       ,      snake_case_	:  Any       ,      snake_case_	:  Optional[int]       ,      snake_case_	:  Union[str, Any]=False       )      ->				List[str]:
   __snake_case       ,	__snake_case							       =   init_clap(snake_case_       ,      enable_fusion=snake_case_       )

   clap_model.eval()
   __snake_case							       =   clap_model.state_dict()
   __snake_case							       =   rename_state_dict(snake_case_       )

   __snake_case							       =   ClapConfig()
   __snake_case							       =   enable_fusion
   __snake_case							       =   ClapModel(snake_case_       )

   # ignore the spectrogram embedding layer
   model.load_state_dict(snake_case_       ,      strict=snake_case_       )

   model.save_pretrained(snake_case_       )
   transformers_config.save_pretrained(snake_case_       )


if __name__ == "__main__":
    snake_case_				=			argparse.ArgumentParser()
    parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
    parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to fairseq checkpoint')
    parser.add_argument('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
    parser.add_argument('--enable_fusion', action='store_true', help='Whether to enable fusion or not')
    snake_case_				=			parser.parse_args()

    convert_clap_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.enable_fusion)








def lowerCamelCase__    (							snake_case_	:  int       )      ->				int:
   if not isinstance(snake_case_       ,      snake_case_       ) or number < 0:
      raise ValueError('''Input must be a non-negative integer'''       )

   __snake_case							       =   0
   while number:
      # This way we arrive at next set bit (next 1) instead of looping
      # through each bit and checking for 1s hence the
      # loop won't run 32 times it will only run the number of `1` times
      number &= number - 1
      count += 1
   return count


if __name__ == "__main__":
    import doctest

    doctest.testmod()

