





from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
       from PIL import Image

       from ..image_utils import load_image

if is_torch_available():
       from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
       from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
       from ..tf_utils import stable_softmax

_A :   Optional[int]  			=  logging.get_logger(__name__)


@add_end_docstrings(lowerCAmelCase_ )
class        __SCREAMING_SNAKE_CASE      (			lowerCAmelCase_ ):







       def __init__(				self					:     Optional[int]		, **A					:     str      )       ->List[str]:
          super().__init__(**A      )

          requires_backends(self		, '''vision'''      )
          self.check_model_type(
              TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == '''tf'''
              else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING      )







       def __call__(				self					:     Dict		, A					:     Union[str, List[str], "Image", List["Image"]]		, **A					:     Optional[Any]      )       ->Optional[Any]:
          return super().__call__(A		, **A      )







       def 						__lowerCamelCase   (				self					:     Optional[Any]		, **A					:     int      )       ->Optional[Any]:
          lowerCamelCase__       :						Optional[Any]        =							{}
          if "candidate_labels" in kwargs:
             lowerCamelCase__       :						Tuple        =							kwargs['''candidate_labels''']
          if "hypothesis_template" in kwargs:
             lowerCamelCase__       :						Optional[Any]        =							kwargs['''hypothesis_template''']

          return preprocess_params, {}, {}







       def 						__lowerCamelCase   (				self					:     Optional[int]		, A					:     Tuple		, A					:     Optional[int]=None		, A					:     Tuple="This is a photo of {}."      )       ->int:
          lowerCamelCase__       :						Any        =							load_image(A      )
          lowerCamelCase__       :						Dict        =							self.image_processor(images=[image]		, return_tensors=self.framework      )
          lowerCamelCase__       :						Union[str, Any]        =							candidate_labels
          lowerCamelCase__       :						List[Any]        =							[hypothesis_template.format(A      ) for x in candidate_labels]
          lowerCamelCase__       :						str        =							self.tokenizer(A		, return_tensors=self.framework		, padding=A      )
          lowerCamelCase__       :						Union[str, Any]        =							[text_inputs]
          return inputs







       def 						__lowerCamelCase   (				self					:     Any		, A					:     Optional[int]      )       ->List[Any]:
          lowerCamelCase__       :						str        =							model_inputs.pop('''candidate_labels'''      )
          lowerCamelCase__       :						Dict        =							model_inputs.pop('''text_inputs'''      )
          if isinstance(text_inputs[0]		, A      ):
             lowerCamelCase__       :						Union[str, Any]        =							text_inputs[0]
          else:
             # Batching case.
             lowerCamelCase__       :						List[Any]        =							text_inputs[0][0]

          lowerCamelCase__       :						int        =							self.model(**A		, **A      )

          lowerCamelCase__       :						Tuple        =							{
              '''candidate_labels''': candidate_labels,
              '''logits''': outputs.logits_per_image,
          }
          return model_outputs







       def 						__lowerCamelCase   (				self					:     int		, A					:     Optional[int]      )       ->Optional[Any]:
          lowerCamelCase__       :						Optional[Any]        =							model_outputs.pop('''candidate_labels'''      )
          lowerCamelCase__       :						Optional[Any]        =							model_outputs['''logits'''][0]
          if self.framework == "pt":
             lowerCamelCase__       :						List[str]        =							logits.softmax(dim=-1      ).squeeze(-1      )
             lowerCamelCase__       :						Dict        =							probs.tolist()
             if not isinstance(A		, A      ):
                lowerCamelCase__       :						Tuple        =							[scores]
          elif self.framework == "tf":
             lowerCamelCase__       :						Union[str, Any]        =							stable_softmax(A		, axis=-1      )
             lowerCamelCase__       :						Optional[int]        =							probs.numpy().tolist()
          else:
             raise ValueError(F"Unsupported framework: {self.framework}"      )

          lowerCamelCase__       :						int        =							[
              {'''score''': score, '''label''': candidate_label}
              for score, candidate_label in sorted(zip(A		, A      )		, key=lambda A      : -x[0]      )
          ]
          return result










import argparse
import re

import torch
from CLAP import create_model

from transformers import AutoFeatureExtractor, ClapConfig, ClapModel


_A :   Optional[Any]  			=  {
    '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',
}

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



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase=False					)    ->    str:







   """simple docstring"""
   lowerCamelCase__       ,							lowerCamelCase__       :						List[str]        =							create_model(
       '''HTSAT-tiny'''      ,  '''roberta'''      ,  UpperCAmelCase      ,  precision='''fp32'''      ,  device='''cuda:0''' if torch.cuda.is_available() else '''cpu'''      ,  enable_fusion=UpperCAmelCase      ,  fusion_type='''aff_2d''' if enable_fusion else None      ,  )
   return model, model_cfg



def 	_a						(   UpperCAmelCase					)    ->    Optional[int]:







   """simple docstring"""
   lowerCamelCase__       :						Optional[int]        =							{}

   lowerCamelCase__       :						int        =							R'''.*sequential.(\d+).*'''
   lowerCamelCase__       :						Any        =							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:
            lowerCamelCase__       :						List[str]        =							key.replace(UpperCAmelCase      ,  UpperCAmelCase					)

      if re.match(UpperCAmelCase      ,  UpperCAmelCase					):
         # replace sequential layers with list
         lowerCamelCase__       :						List[Any]        =							re.match(UpperCAmelCase      ,  UpperCAmelCase					).group(1					)

         lowerCamelCase__       :						Optional[int]        =							key.replace(f"sequential.{sequential_layer}."      ,  f"layers.{int(UpperCAmelCase					)//3}.linear."					)
      elif re.match(UpperCAmelCase      ,  UpperCAmelCase					):
         lowerCamelCase__       :						Optional[int]        =							int(re.match(UpperCAmelCase      ,  UpperCAmelCase					).group(1					)					)

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

         lowerCamelCase__       :						List[str]        =							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
         lowerCamelCase__       :						Optional[Any]        =							value
         lowerCamelCase__       :						Optional[Any]        =							mixed_qkv.size(0					) // 3

         lowerCamelCase__       :						Tuple        =							mixed_qkv[:qkv_dim]
         lowerCamelCase__       :						Dict        =							mixed_qkv[qkv_dim : qkv_dim * 2]
         lowerCamelCase__       :						int        =							mixed_qkv[qkv_dim * 2 :]

         lowerCamelCase__       :						Optional[int]        =							query_layer
         lowerCamelCase__       :						str        =							key_layer
         lowerCamelCase__       :						List[str]        =							value_layer
      else:
         lowerCamelCase__       :						Tuple        =							value

   return model_state_dict



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase=False					)    ->    List[Any]:







   """simple docstring"""
   lowerCamelCase__       ,							lowerCamelCase__       :						Dict        =							init_clap(UpperCAmelCase      ,  enable_fusion=UpperCAmelCase					)

   clap_model.eval()
   lowerCamelCase__       :						List[Any]        =							clap_model.state_dict()
   lowerCamelCase__       :						Dict        =							rename_state_dict(UpperCAmelCase					)

   lowerCamelCase__       :						Optional[int]        =							ClapConfig()
   lowerCamelCase__       :						Optional[int]        =							enable_fusion
   lowerCamelCase__       :						Optional[Any]        =							ClapModel(UpperCAmelCase					)

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

   model.save_pretrained(UpperCAmelCase					)
   transformers_config.save_pretrained(UpperCAmelCase					)


if __name__ == "__main__":
       _A :   int  			=  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')
       _A :   Tuple  			=  parser.parse_args()

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



