




# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool




class 		__A      (			UpperCamelCase__			):
						UpperCamelCase     						=					"""facebook/bart-large-mnli"""
						UpperCamelCase     						=					(
						    """This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which """
						    """should be the text to classify, and `labels`, which should be the list of labels to use for classification. """
						    """It returns the most likely label in the list of provided `labels` for the input text."""
						)
						UpperCamelCase     						=					"""text_classifier"""
						UpperCamelCase     						=					AutoTokenizer
						UpperCamelCase     						=					AutoModelForSequenceClassification

						UpperCamelCase     						=					["""text""", ["""text"""]]
						UpperCamelCase     						=					["""text"""]

						def     A__						(		self   :Optional[Any] ):
										'''simple docstring'''

										super().setup()
										__magic_name__	:	Union[str, Any]								=self.model.config
										__magic_name__	:	Union[str, Any]								=-1
										for idx, label in config.idalabel.items():
														if label.lower().startswith("""entail""" ):
																		__magic_name__	:	Optional[Any]								=int(__snake_case )
										if self.entailment_id == -1:
														raise ValueError("""Could not determine the entailment ID from the model config, please pass it at init.""" )

						def     A__						(		self   :Any    , __snake_case   :Tuple    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	str								=labels
										return self.pre_processor(
										    [text] * len(__snake_case )    , [f"This example is {label}" for label in labels]    , return_tensors="""pt"""    , padding="""max_length"""    , )


						def     A__						(		self   :Dict    , __snake_case   :str ):
										'''simple docstring'''

										__magic_name__	:	Dict								=outputs.logits
										__magic_name__	:	List[Any]								=torch.argmax(logits[:, 2] ).item()
										return self._labels[label_id]










import argparse
import re

import torch
from CLAP import create_model

from transformers import AutoFeatureExtractor, ClapConfig, ClapModel


UpperCAmelCase_		:				Optional[int]									=     {
    "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",
}

UpperCAmelCase_		:				Optional[int]									=     AutoFeatureExtractor.from_pretrained("laion/clap-htsat-unfused", truncation="rand_trunc")




def 		lowerCAmelCase_   (	lowerCamelCase	,     lowerCamelCase=False   ):
				__magic_name__				, __magic_name__	:	Any								=create_model(
				    """HTSAT-tiny"""	,     """roberta"""	,     lowerCamelCase	,     precision="""fp32"""	,     device="""cuda:0""" if torch.cuda.is_available() else """cpu"""	,     enable_fusion=lowerCamelCase	,     fusion_type="""aff_2d""" if enable_fusion else None	,     )
				return model, model_cfg




def 		lowerCAmelCase_   (	lowerCamelCase   ):
				__magic_name__	:	Optional[int]								={}

				__magic_name__	:	Union[str, Any]								=R""".*sequential.(\d+).*"""
				__magic_name__	:	Union[str, 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:
																__magic_name__	:	Union[str, Any]								=key.replace(lowerCamelCase	,     lowerCamelCase   )

								if re.match(lowerCamelCase	,     lowerCamelCase   ):
												# replace sequential layers with list
												__magic_name__	:	str								=re.match(lowerCamelCase	,     lowerCamelCase   ).group(1   )

												__magic_name__	:	Optional[Any]								=key.replace(F"sequential.{sequential_layer}."	,     F"layers.{int(lowerCamelCase   )//3}.linear."   )
								elif re.match(lowerCamelCase	,     lowerCamelCase   ):
												__magic_name__	:	Any								=int(re.match(lowerCamelCase	,     lowerCamelCase   ).group(1   )   )

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

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

												__magic_name__	:	Optional[Any]								=mixed_qkv[:qkv_dim]
												__magic_name__	:	Any								=mixed_qkv[qkv_dim : qkv_dim * 2]
												__magic_name__	:	Optional[Any]								=mixed_qkv[qkv_dim * 2 :]

												__magic_name__	:	str								=query_layer
												__magic_name__	:	int								=key_layer
												__magic_name__	:	str								=value_layer
								else:
												__magic_name__	:	int								=value

				return model_state_dict




def 		lowerCAmelCase_   (	lowerCamelCase	,     lowerCamelCase	,     lowerCamelCase	,     lowerCamelCase=False   ):
				__magic_name__				, __magic_name__	:	Tuple								=init_clap(lowerCamelCase	,     enable_fusion=lowerCamelCase   )

				clap_model.eval()
				__magic_name__	:	int								=clap_model.state_dict()
				__magic_name__	:	Optional[Any]								=rename_state_dict(lowerCamelCase   )

				__magic_name__	:	Any								=ClapConfig()
				__magic_name__	:	Optional[Any]								=enable_fusion
				__magic_name__	:	Tuple								=ClapModel(lowerCamelCase   )

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

				model.save_pretrained(lowerCamelCase   )
				transformers_config.save_pretrained(lowerCamelCase   )


if __name__ == "__main__":
							UpperCAmelCase_		:				Optional[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")
							UpperCAmelCase_		:				Dict									=     parser.parse_args()

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




