





import argparse

import torch

from transformers import (
    UniSpeechSatConfig,
    UniSpeechSatForAudioFrameClassification,
    UniSpeechSatForSequenceClassification,
    UniSpeechSatForXVector,
    WavaVecaFeatureExtractor,
    logging,
)


logging.set_verbosity_info()
_lowerCamelCase 					=      logging.get_logger(__name__)






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				List[Any]       , UpperCamelCase__:				Tuple       , UpperCamelCase__:				List[Any]     ):
		SCREAMING_SNAKE_CASE__								=   UniSpeechSatForSequenceClassification.from_pretrained(UpperCamelCase__       , config=UpperCamelCase__     )
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""projector.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""projector.bias"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.post_net.linear.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.post_net.linear.bias"""]
		return model






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				List[Any]       , UpperCamelCase__:				int       , UpperCamelCase__:				int     ):
		SCREAMING_SNAKE_CASE__								=   UniSpeechSatForAudioFrameClassification.from_pretrained(UpperCamelCase__       , config=UpperCamelCase__     )
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.linear.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.linear.bias"""]
		return model






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				List[str]       , UpperCamelCase__:				Optional[int]       , UpperCamelCase__:				Dict     ):
		SCREAMING_SNAKE_CASE__								=   UniSpeechSatForXVector.from_pretrained(UpperCamelCase__       , config=UpperCamelCase__     )
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""connector.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""connector.bias"""]
		for i, kernel_size in enumerate(hf_config.tdnn_kernel     ):
				SCREAMING_SNAKE_CASE__								=   downstream_dict[
				    f'''model.framelevel_feature_extractor.module.{i}.kernel.weight'''
				]
				SCREAMING_SNAKE_CASE__								=   downstream_dict[f'''model.framelevel_feature_extractor.module.{i}.kernel.bias''']

		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.utterancelevel_feature_extractor.linear1.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.utterancelevel_feature_extractor.linear1.bias"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.utterancelevel_feature_extractor.linear2.weight"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""model.utterancelevel_feature_extractor.linear2.bias"""]
		SCREAMING_SNAKE_CASE__								=   downstream_dict["""objective.W"""]
		return model






@torch.no_grad()
def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				str       , UpperCamelCase__:				Any       , UpperCamelCase__:				Dict       , UpperCamelCase__:				Optional[Any]     ):
		SCREAMING_SNAKE_CASE__								=   torch.load(UpperCamelCase__       , map_location="""cpu"""     )

		SCREAMING_SNAKE_CASE__								=   checkpoint["""Downstream"""]

		SCREAMING_SNAKE_CASE__								=   UniSpeechSatConfig.from_pretrained(UpperCamelCase__     )
		SCREAMING_SNAKE_CASE__								=   WavaVecaFeatureExtractor.from_pretrained(
		    UpperCamelCase__       , return_attention_mask=UpperCamelCase__       , do_normalize=UpperCamelCase__     )

		SCREAMING_SNAKE_CASE__								=   hf_config.architectures[0]
		if arch.endswith("""ForSequenceClassification"""     ):
				SCREAMING_SNAKE_CASE__								=   convert_classification(UpperCamelCase__       , UpperCamelCase__       , UpperCamelCase__     )
		elif arch.endswith("""ForAudioFrameClassification"""     ):
				SCREAMING_SNAKE_CASE__								=   convert_diarization(UpperCamelCase__       , UpperCamelCase__       , UpperCamelCase__     )
		elif arch.endswith("""ForXVector"""     ):
				SCREAMING_SNAKE_CASE__								=   convert_xvector(UpperCamelCase__       , UpperCamelCase__       , UpperCamelCase__     )
		else:
				raise NotImplementedError(f'''S3PRL weights conversion is not supported for {arch}'''     )

		if hf_config.use_weighted_layer_sum:
				SCREAMING_SNAKE_CASE__								=   checkpoint["""Featurizer"""]["""weights"""]

		hf_feature_extractor.save_pretrained(UpperCamelCase__     )
		hf_model.save_pretrained(UpperCamelCase__     )


if __name__ == "__main__":
	_lowerCamelCase 					=      argparse.ArgumentParser()
	parser.add_argument(
	    '--base_model_name', default=None, type=str, help='Name of the huggingface pretrained base model.'
	)
	parser.add_argument('--config_path', default=None, type=str, help='Path to the huggingface classifier config.')
	parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to the s3prl checkpoint.')
	parser.add_argument('--model_dump_path', default=None, type=str, help='Path to the final converted model.')
	_lowerCamelCase 					=      parser.parse_args()
	convert_saprl_checkpoint(args.base_model_name, args.config_path, args.checkpoint_path, args.model_dump_path)






import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible




class 	UpperCamelCase_							(  unittest.TestCase     ):


							def   _snake_case	(				self     :Tuple		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.bin""",
									    """safety_checker/model.safetensors""",
									    """vae/diffusion_pytorch_model.bin""",
									    """vae/diffusion_pytorch_model.safetensors""",
									    """text_encoder/pytorch_model.bin""",
									    """text_encoder/model.safetensors""",
									    """unet/diffusion_pytorch_model.bin""",
									    """unet/diffusion_pytorch_model.safetensors""",
									]
									self.assertTrue(is_safetensors_compatible(__A		)		)


							def   _snake_case	(				self     :List[str]		) -> int:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """unet/diffusion_pytorch_model.bin""",
									    """unet/diffusion_pytorch_model.safetensors""",
									]
									self.assertTrue(is_safetensors_compatible(__A		)		)


							def   _snake_case	(				self     :Tuple		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.bin""",
									    """safety_checker/model.safetensors""",
									    """vae/diffusion_pytorch_model.bin""",
									    """vae/diffusion_pytorch_model.safetensors""",
									    """text_encoder/pytorch_model.bin""",
									    """text_encoder/model.safetensors""",
									    """unet/diffusion_pytorch_model.bin""",
									    # Removed: 'unet/diffusion_pytorch_model.safetensors',
									]
									self.assertFalse(is_safetensors_compatible(__A		)		)


							def   _snake_case	(				self     :Union[str, Any]		) -> Union[str, Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """text_encoder/pytorch_model.bin""",
									    """text_encoder/model.safetensors""",
									]
									self.assertTrue(is_safetensors_compatible(__A		)		)


							def   _snake_case	(				self     :Optional[Any]		) -> Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.bin""",
									    """safety_checker/model.safetensors""",
									    """vae/diffusion_pytorch_model.bin""",
									    """vae/diffusion_pytorch_model.safetensors""",
									    """text_encoder/pytorch_model.bin""",
									    # Removed: 'text_encoder/model.safetensors',
									    """unet/diffusion_pytorch_model.bin""",
									    """unet/diffusion_pytorch_model.safetensors""",
									]
									self.assertFalse(is_safetensors_compatible(__A		)		)


							def   _snake_case	(				self     :Tuple		) -> List[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.fp16.bin""",
									    """safety_checker/model.fp16.safetensors""",
									    """vae/diffusion_pytorch_model.fp16.bin""",
									    """vae/diffusion_pytorch_model.fp16.safetensors""",
									    """text_encoder/pytorch_model.fp16.bin""",
									    """text_encoder/model.fp16.safetensors""",
									    """unet/diffusion_pytorch_model.fp16.bin""",
									    """unet/diffusion_pytorch_model.fp16.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertTrue(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :Any		) -> Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """unet/diffusion_pytorch_model.fp16.bin""",
									    """unet/diffusion_pytorch_model.fp16.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertTrue(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :str		) -> Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """unet/diffusion_pytorch_model.bin""",
									    """unet/diffusion_pytorch_model.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertTrue(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :List[Any]		) -> List[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.fp16.bin""",
									    """safety_checker/model.fp16.safetensors""",
									    """vae/diffusion_pytorch_model.fp16.bin""",
									    """vae/diffusion_pytorch_model.fp16.safetensors""",
									    """text_encoder/pytorch_model.fp16.bin""",
									    """text_encoder/model.fp16.safetensors""",
									    """unet/diffusion_pytorch_model.fp16.bin""",
									    # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertFalse(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :str		) -> List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """text_encoder/pytorch_model.fp16.bin""",
									    """text_encoder/model.fp16.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertTrue(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :Optional[int]		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """text_encoder/pytorch_model.bin""",
									    """text_encoder/model.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertTrue(is_safetensors_compatible(__A				,					variant=__A		)		)


							def   _snake_case	(				self     :Union[str, Any]		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   [
									    """safety_checker/pytorch_model.fp16.bin""",
									    """safety_checker/model.fp16.safetensors""",
									    """vae/diffusion_pytorch_model.fp16.bin""",
									    """vae/diffusion_pytorch_model.fp16.safetensors""",
									    """text_encoder/pytorch_model.fp16.bin""",
									    # 'text_encoder/model.fp16.safetensors',
									    """unet/diffusion_pytorch_model.fp16.bin""",
									    """unet/diffusion_pytorch_model.fp16.safetensors""",
									]
									SCREAMING_SNAKE_CASE__								=   """fp16"""
									self.assertFalse(is_safetensors_compatible(__A				,					variant=__A		)		)