



'''simple docstring'''




import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConformerConfig,
    WavaVecaConformerForCTC,
    WavaVecaConformerForPreTraining,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
lowerCamelCase							:Tuple       = logging.get_logger(__name__)

lowerCamelCase							:Any       = {
    '''post_extract_proj''': '''feature_projection.projection''',
    '''encoder.pos_conv.0''': '''encoder.pos_conv_embed.conv''',
    '''self_attn.linear_k''': '''encoder.layers.*.self_attn.linear_k''',
    '''self_attn.linear_v''': '''encoder.layers.*.self_attn.linear_v''',
    '''self_attn.linear_q''': '''encoder.layers.*.self_attn.linear_q''',
    '''self_attn.pos_bias_u''': '''encoder.layers.*.self_attn.pos_bias_u''',
    '''self_attn.pos_bias_v''': '''encoder.layers.*.self_attn.pos_bias_v''',
    '''self_attn.linear_out''': '''encoder.layers.*.self_attn.linear_out''',
    '''self_attn.linear_pos''': '''encoder.layers.*.self_attn.linear_pos''',
    '''self_attn.rotary_emb''': '''encoder.embed_positions''',
    '''self_attn_layer_norm''': '''encoder.layers.*.self_attn_layer_norm''',
    '''conv_module.pointwise_conv1''': '''encoder.layers.*.conv_module.pointwise_conv1''',
    '''conv_module.pointwise_conv2''': '''encoder.layers.*.conv_module.pointwise_conv2''',
    '''conv_module.depthwise_conv''': '''encoder.layers.*.conv_module.depthwise_conv''',
    '''conv_module.batch_norm''': '''encoder.layers.*.conv_module.batch_norm''',
    '''conv_module.layer_norm''': '''encoder.layers.*.conv_module.layer_norm''',
    '''ffn1.w_1''': '''encoder.layers.*.ffn1.intermediate_dense''',
    '''ffn1.w_2''': '''encoder.layers.*.ffn1.output_dense''',
    '''ffn1.layer_norm''': '''encoder.layers.*.ffn1_layer_norm''',
    '''ffn2.w_1''': '''encoder.layers.*.ffn2.intermediate_dense''',
    '''ffn2.w_2''': '''encoder.layers.*.ffn2.output_dense''',
    '''ffn2.layer_norm''': '''encoder.layers.*.ffn2_layer_norm''',
    '''final_layer_norm''': '''encoder.layers.*.final_layer_norm''',
    '''encoder.layer_norm''': '''encoder.layer_norm''',
    '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
    '''quantizer.weight_proj''': '''quantizer.weight_proj''',
    '''quantizer.vars''': '''quantizer.codevectors''',
    '''project_q''': '''project_q''',
    '''final_proj''': '''project_hid''',
    '''w2v_encoder.proj''': '''lm_head''',
    '''mask_emb''': '''masked_spec_embed''',
}
lowerCamelCase							:int       = [
    '''lm_head''',
    '''quantizer.weight_proj''',
    '''quantizer.codevectors''',
    '''project_q''',
    '''project_hid''',
]



def      a      (			lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__   ):
			'''simple docstring'''





			for attribute in key.split("""."""   ):
						A_     :						str     =					getattr(lowerCamelCase__						,				lowerCamelCase__   )

			if weight_type is not None:
						A_     :						Tuple     =					getattr(lowerCamelCase__						,				lowerCamelCase__   ).shape
			else:
						A_     :						Any     =					hf_pointer.shape

			if hf_shape != value.shape:
						raise ValueError(
						    f'Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be'
						    f' {value.shape} for {full_name}'   )

			if weight_type == "weight":
						A_     :						Any     =					value
			elif weight_type == "weight_g":
						A_     :						List[Any]     =					value
			elif weight_type == "weight_v":
						A_     :						List[str]     =					value
			elif weight_type == "bias":
						A_     :						Union[str, Any]     =					value
			elif weight_type == "running_mean":
						A_     :						List[str]     =					value
			elif weight_type == "running_var":
						A_     :						Optional[Any]     =					value
			elif weight_type == "num_batches_tracked":
						A_     :						Tuple     =					value
			elif weight_type == "inv_freq":
						A_     :						Dict     =					value
			else:
						A_     :						Optional[Any]     =					value

			logger.info(f'{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'   )



def      a      (			lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__   ):
			'''simple docstring'''





			A_     :						List[Any]     =					[]
			A_     :						Tuple     =					fairseq_model.state_dict()

			A_     :						Optional[Any]     =					hf_model.wavaveca_conformer.feature_extractor

			for name, value in fairseq_dict.items():
						A_     :						Union[str, Any]     =					False
						if "conv_layers" in name:
									load_conv_layer(
									    lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				hf_model.config.feat_extract_norm == """group"""						,				)
									A_     :						List[str]     =					True
						else:
									for key, mapped_key in MAPPING.items():
												A_     :						List[str]     =					"""wav2vec2_conformer.""" + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
												if key in name or key.split("""w2v_model."""   )[-1] == name.split("""."""   )[0]:
															A_     :						str     =					True
															if "*" in mapped_key:
																		A_     :						Optional[int]     =					name.split(lowerCamelCase__   )[0].split("""."""   )[-2]
																		A_     :						Optional[Any]     =					mapped_key.replace("""*"""						,				lowerCamelCase__   )
															if "pos_bias_u" in name:
																		A_     :						Optional[Any]     =					None
															elif "pos_bias_v" in name:
																		A_     :						Tuple     =					None
															elif "weight_g" in name:
																		A_     :						List[str]     =					"""weight_g"""
															elif "weight_v" in name:
																		A_     :						int     =					"""weight_v"""
															elif "bias" in name:
																		A_     :						List[str]     =					"""bias"""
															elif "weight" in name:
																		# TODO: don't match quantizer.weight_proj
																		A_     :						Any     =					"""weight"""
															elif "running_mean" in name:
																		A_     :						Optional[Any]     =					"""running_mean"""
															elif "inv_freq" in name:
																		A_     :						Tuple     =					"""inv_freq"""
															elif "running_var" in name:
																		A_     :						Dict     =					"""running_var"""
															elif "num_batches_tracked" in name:
																		A_     :						str     =					"""num_batches_tracked"""
															else:
																		A_     :						Tuple     =					None
															set_recursively(lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__   )
												continue
						if not is_used:
									unused_weights.append(lowerCamelCase__   )

			logger.warning(f'Unused weights: {unused_weights}'   )



def      a      (			lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__   ):
			'''simple docstring'''





			A_     :						Dict     =					full_name.split("""conv_layers."""   )[-1]
			A_     :						int     =					name.split("""."""   )
			A_     :						Optional[int]     =					int(items[0]   )
			A_     :						Optional[Any]     =					int(items[1]   )

			if type_id == 0:
						if "bias" in name:
									if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
												raise ValueError(
												    f'{full_name} has size {value.shape}, but'
												    f' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'   )
									A_     :						Optional[int]     =					value
									logger.info(f'Feat extract conv layer {layer_id} was initialized from {full_name}.'   )
						elif "weight" in name:
									if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
												raise ValueError(
												    f'{full_name} has size {value.shape}, but'
												    f' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'   )
									A_     :						Optional[Any]     =					value
									logger.info(f'Feat extract conv layer {layer_id} was initialized from {full_name}.'   )
			elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
						if "bias" in name:
									if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
												raise ValueError(
												    f'{full_name} has size {value.shape}, but'
												    f' {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found.'   )
									A_     :						Dict     =					value
									logger.info(f'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'   )
						elif "weight" in name:
									if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
												raise ValueError(
												    f'{full_name} has size {value.shape}, but'
												    f' {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found.'   )
									A_     :						Union[str, Any]     =					value
									logger.info(f'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'   )
			else:
						unused_weights.append(lowerCamelCase__   )



@torch.no_grad()
def      a      (			lowerCamelCase__						,				lowerCamelCase__						,				lowerCamelCase__=None						,				lowerCamelCase__=None						,				lowerCamelCase__=True   ):
			'''simple docstring'''





			if config_path is not None:
						A_     :						Union[str, Any]     =					WavaVecaConformerConfig.from_pretrained(lowerCamelCase__						,				hidden_act="""swish"""   )
			else:
						A_     :						Tuple     =					WavaVecaConformerConfig()

			if "rope" in checkpoint_path:
						A_     :						List[str]     =					"""rotary"""

			if is_finetuned:
						if dict_path:
									A_     :						int     =					Dictionary.load(lowerCamelCase__   )

									# important change bos & pad token id since CTC symbol is <pad> and
									# not <s> as in fairseq
									A_     :						Dict     =					target_dict.pad_index
									A_     :						List[Any]     =					target_dict.bos_index
									A_     :						Any     =					target_dict.eos_index
									A_     :						Optional[int]     =					len(target_dict.symbols   )
									A_     :						int     =					os.path.join(lowerCamelCase__						,				"""vocab.json"""   )
									if not os.path.isdir(lowerCamelCase__   ):
												logger.error("""--pytorch_dump_folder_path ({}) should be a directory""".format(lowerCamelCase__   )   )
												return
									os.makedirs(lowerCamelCase__						,				exist_ok=lowerCamelCase__   )
									A_     :						int     =					target_dict.indices

									# fairseq has the <pad> and <s> switched
									A_     :						List[Any]     =					0
									A_     :						Tuple     =					1
									with open(lowerCamelCase__						,				"""w"""						,				encoding="""utf-8"""   ) as vocab_handle:
												json.dump(lowerCamelCase__						,				lowerCamelCase__   )
									A_     :						Tuple     =					WavaVecaCTCTokenizer(
									    lowerCamelCase__						,				unk_token=target_dict.unk_word						,				pad_token=target_dict.pad_word						,				bos_token=target_dict.bos_word						,				eos_token=target_dict.eos_word						,				word_delimiter_token="""|"""						,				do_lower_case=lowerCamelCase__						,				)
									A_     :						int     =					True if config.feat_extract_norm == """layer""" else False
									A_     :						List[str]     =					WavaVecaFeatureExtractor(
									    feature_size=1						,				sampling_rate=1_60_00						,				padding_value=0						,				do_normalize=lowerCamelCase__						,				return_attention_mask=lowerCamelCase__						,				)
									A_     :						Optional[Any]     =					WavaVecaProcessor(feature_extractor=lowerCamelCase__						,				tokenizer=lowerCamelCase__   )
									processor.save_pretrained(lowerCamelCase__   )

						A_     :						str     =					WavaVecaConformerForCTC(lowerCamelCase__   )
			else:
						A_     :						Dict     =					WavaVecaConformerForPreTraining(lowerCamelCase__   )

			if is_finetuned:
						A_,   A_,   A_     :						List[str]     =					fairseq.checkpoint_utils.load_model_ensemble_and_task(
						    [checkpoint_path]						,				arg_overrides={"""data""": """/""".join(dict_path.split("""/"""   )[:-1]   )}   )
			else:
						A_     :						Union[str, Any]     =					argparse.Namespace(task="""audio_pretraining"""   )
						A_     :						Optional[Any]     =					fairseq.tasks.setup_task(lowerCamelCase__   )

						A_,   A_,   A_     :						Any     =					fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]						,				task=lowerCamelCase__   )

			A_     :						List[str]     =					model[0].eval()

			recursively_load_weights(lowerCamelCase__						,				lowerCamelCase__						,				not is_finetuned   )

			hf_wavavec.save_pretrained(lowerCamelCase__   )


if __name__ == "__main__":
						lowerCamelCase							: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('''--dict_path''', default=None, type=str, help='''Path to dict of fine-tuned model''')
						parser.add_argument('''--config_path''', default=None, type=str, help='''Path to hf config.json of model to convert''')
						parser.add_argument(
						    '''--not_finetuned''', action='''store_true''', help='''Whether the model to convert is a fine-tuned model or not'''
						)
						lowerCamelCase							:Union[str, Any]       = parser.parse_args()
						convert_wavaveca_conformer_checkpoint(
						    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
						)




'''simple docstring'''




from __future__ import annotations

from functools import lru_cache
from math import ceil

lowerCamelCase							:List[str]       = 1_0_0

lowerCamelCase							:Dict       = set(range(3, NUM_PRIMES, 2))
primes.add(2)
lowerCamelCase							:int

for prime in range(3, ceil(NUM_PRIMES**0.5), 2):
						if prime not in primes:
												continue
						primes.difference_update(set(range(prime * prime, NUM_PRIMES, prime)))



@lru_cache(maxsize=1_00   )
def      a      (			lowerCamelCase__   ):
			'''simple docstring'''





			if number_to_partition < 0:
						return set()
			elif number_to_partition == 0:
						return {1}

			A_     :						set[int]     =					set()
			A_     :						int
			A_     :						int

			for prime in primes:
						if prime > number_to_partition:
									continue
						for sub in partition(number_to_partition - prime   ):
									ret.add(sub * prime   )

			return ret



def      a      (			lowerCamelCase__ = 50_00   ):
			'''simple docstring'''





			for number_to_partition in range(1						,				lowerCamelCase__   ):
						if len(partition(lowerCamelCase__   )   ) > number_unique_partitions:
									return number_to_partition
			return None


if __name__ == "__main__":
						print(F"{solution() = }")