





import argparse

import torch

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


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




def     lowerCamelCase_	(      lowerCAmelCase:	Tuple    ,							lowerCAmelCase:	Optional[Any]    ,							lowerCAmelCase:	Tuple				)->		List[Any]:
  _snake_case   :				List[str]														=	UniSpeechSatForSequenceClassification.from_pretrained(lowerCamelCase_    ,							config=lowerCamelCase_				)
  _snake_case   :				Tuple														=	downstream_dict['projector.weight']
  _snake_case   :				Any														=	downstream_dict['projector.bias']
  _snake_case   :				Dict														=	downstream_dict['model.post_net.linear.weight']
  _snake_case   :				int														=	downstream_dict['model.post_net.linear.bias']
  return model




def     lowerCamelCase_	(      lowerCAmelCase:	Optional[int]    ,							lowerCAmelCase:	Dict    ,							lowerCAmelCase:	Any				)->		int:
  _snake_case   :				Dict														=	UniSpeechSatForAudioFrameClassification.from_pretrained(lowerCamelCase_    ,							config=lowerCamelCase_				)
  _snake_case   :				List[str]														=	downstream_dict['model.linear.weight']
  _snake_case   :				Optional[int]														=	downstream_dict['model.linear.bias']
  return model




def     lowerCamelCase_	(      lowerCAmelCase:	int    ,							lowerCAmelCase:	Any    ,							lowerCAmelCase:	Union[str, Any]				)->		Tuple:
  _snake_case   :				Tuple														=	UniSpeechSatForXVector.from_pretrained(lowerCamelCase_    ,							config=lowerCamelCase_				)
  _snake_case   :				Tuple														=	downstream_dict['connector.weight']
  _snake_case   :				str														=	downstream_dict['connector.bias']
  for i, kernel_size in enumerate(hf_config.tdnn_kernel				):
    _snake_case   :				Tuple														=	downstream_dict[
        F"""model.framelevel_feature_extractor.module.{i}.kernel.weight"""
    ]
    _snake_case   :				str														=	downstream_dict[F"""model.framelevel_feature_extractor.module.{i}.kernel.bias"""]

  _snake_case   :				List[Any]														=	downstream_dict['model.utterancelevel_feature_extractor.linear1.weight']
  _snake_case   :				Optional[Any]														=	downstream_dict['model.utterancelevel_feature_extractor.linear1.bias']
  _snake_case   :				Union[str, Any]														=	downstream_dict['model.utterancelevel_feature_extractor.linear2.weight']
  _snake_case   :				Union[str, Any]														=	downstream_dict['model.utterancelevel_feature_extractor.linear2.bias']
  _snake_case   :				int														=	downstream_dict['objective.W']
  return model




@torch.no_grad()
def     lowerCamelCase_	(      lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	List[str]    ,							lowerCAmelCase:	Any    ,							lowerCAmelCase:	Union[str, Any]				)->		Dict:
  _snake_case   :				List[Any]														=	torch.load(lowerCamelCase_    ,							map_location='cpu'				)

  _snake_case   :				int														=	checkpoint['Downstream']

  _snake_case   :				Union[str, Any]														=	UniSpeechSatConfig.from_pretrained(lowerCamelCase_				)
  _snake_case   :				List[str]														=	WavaVecaFeatureExtractor.from_pretrained(
      lowerCamelCase_    ,							return_attention_mask=lowerCamelCase_    ,							do_normalize=lowerCamelCase_				)

  _snake_case   :				Optional[Any]														=	hf_config.architectures[0]
  if arch.endswith('ForSequenceClassification'				):
    _snake_case   :				List[str]														=	convert_classification(lowerCamelCase_    ,							lowerCamelCase_    ,							lowerCamelCase_				)
  elif arch.endswith('ForAudioFrameClassification'				):
    _snake_case   :				List[str]														=	convert_diarization(lowerCamelCase_    ,							lowerCamelCase_    ,							lowerCamelCase_				)
  elif arch.endswith('ForXVector'				):
    _snake_case   :				Optional[int]														=	convert_xvector(lowerCamelCase_    ,							lowerCamelCase_    ,							lowerCamelCase_				)
  else:
    raise NotImplementedError(F"""S3PRL weights conversion is not supported for {arch}"""				)

  if hf_config.use_weighted_layer_sum:
    _snake_case   :				Optional[int]														=	checkpoint['Featurizer']['weights']

  hf_feature_extractor.save_pretrained(lowerCamelCase_				)
  hf_model.save_pretrained(lowerCamelCase_				)


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)











def     lowerCamelCase_	(      lowerCAmelCase:	bytes				)->		str:
  return "".join([hex(lowerCAmelCase				)[2:].zfill(2				).upper() for byte in list(lowerCAmelCase				)]				)


def     lowerCamelCase_	(      lowerCAmelCase:	str				)->		bytes:
  # Check data validity, following RFC3548
  # https://www.ietf.org/rfc/rfc3548.txt
  if (len(lowerCAmelCase				) % 2) != 0:
    raise ValueError(
        'Base16 encoded data is invalid:\nData does not have an even number of hex digits.'				)
  # Check the character set - the standard base16 alphabet
  # is uppercase according to RFC3548 section 6
  if not set(lowerCAmelCase				) <= set('0123456789ABCDEF'				):
    raise ValueError(
        'Base16 encoded data is invalid:\nData is not uppercase hex or it contains invalid characters.'				)
  # For every two hexadecimal digits (= a byte), turn it into an integer.
  # Then, string the result together into bytes, and return it.
  return bytes(int(data[i] + data[i + 1]    ,							16				) for i in range(0    ,							len(lowerCAmelCase				)    ,							2				)				)


if __name__ == "__main__":
      import doctest

      doctest.testmod()




