






'''simple docstring'''




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__lowerCamelCase   :					str   							=					logging.get_logger(__name__)







class    A_	(a_							):
							"""simple docstring"""




							a__   		=	['''input_features''', '''attention_mask''']

							def __init__(		self			:Optional[Any]       ,							lowerCAmelCase__			:str=80       ,							lowerCAmelCase__			:Optional[Any]=16_000       ,							lowerCAmelCase__			:List[str]=80       ,							lowerCAmelCase__			:str=0.0       ,							lowerCAmelCase__			:int=True       ,							lowerCAmelCase__			:int=True       ,							lowerCAmelCase__			:Union[str, Any]=True       ,							**lowerCAmelCase__			:Tuple       ,							)				->  List[str]:


										'''simple docstring'''




										super().__init__(feature_size=lowerCAmelCase__       ,							sampling_rate=lowerCAmelCase__       ,							padding_value=lowerCAmelCase__       ,							**lowerCAmelCase__				)
										snake_case_				:					List[str]   =       num_mel_bins
										snake_case_				:					Any   =       do_ceptral_normalize
										snake_case_				:					Tuple   =       normalize_means
										snake_case_				:					Optional[int]   =       normalize_vars
										snake_case_				:					Dict   =       True

							def 							_A	(		self			:List[Any]       ,							lowerCAmelCase__			:np.ndarray       ,							)				->  np.ndarray:


										'''simple docstring'''




										snake_case_				:					Optional[int]   =       waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
										snake_case_				:					Dict   =       torch.from_numpy(lowerCAmelCase__				).unsqueeze(0				)
										snake_case_				:					Union[str, Any]   =       ta_kaldi.fbank(lowerCAmelCase__       ,							num_mel_bins=self.num_mel_bins       ,							sample_frequency=self.sampling_rate				)
										return features.numpy()

							@staticmethod
							def 							_A	(		lowerCAmelCase__			:np.ndarray       ,							lowerCAmelCase__			:int       ,							lowerCAmelCase__			:Optional[bool] = True       ,							lowerCAmelCase__			:Optional[bool] = True       ,							lowerCAmelCase__			:float = 0.0       ,							)				->  np.ndarray:


										'''simple docstring'''




										if normalize_means:
													snake_case_				:					int   =       x[:input_length].mean(axis=0				)
													snake_case_				:					List[str]   =       np.subtract(lowerCAmelCase__       ,							lowerCAmelCase__				)
										if normalize_vars:
													snake_case_				:					Union[str, Any]   =       x[:input_length].std(axis=0				)
													snake_case_				:					int   =       np.divide(lowerCAmelCase__       ,							lowerCAmelCase__				)

										if input_length < x.shape[0]:
													snake_case_				:					Union[str, Any]   =       padding_value

										# make sure array is in float32
										snake_case_				:					Union[str, Any]   =       x.astype(np.floataa				)

										return x

							def 							_A	(		self			:Union[str, Any]       ,							lowerCAmelCase__			:List[np.ndarray]       ,							lowerCAmelCase__			:Optional[np.ndarray] = None				)				->  List[np.ndarray]:


										'''simple docstring'''




										snake_case_				:					int   =       attention_mask.sum(-1				) if attention_mask is not None else [x.shape[0] for x in input_features]
										return [
										    self.utterance_cmvn(lowerCAmelCase__       ,							lowerCAmelCase__       ,							self.normalize_means       ,							self.normalize_vars       ,							self.padding_value				)
										    for x, n in zip(lowerCAmelCase__       ,							lowerCAmelCase__				)
										]






							def __call__(		self			:Union[str, Any]       ,							lowerCAmelCase__			:Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,							lowerCAmelCase__			:Union[bool, str, PaddingStrategy] = False       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:bool = False       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:Optional[Union[str, TensorType]] = None       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:Optional[bool] = None       ,							**lowerCAmelCase__			:Optional[int]       ,							)				->  BatchFeature:


										'''simple docstring'''




										if sampling_rate is not None:
													if sampling_rate != self.sampling_rate:
																raise ValueError(
																    F'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
																    F''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
																    F''' {self.sampling_rate} and not {sampling_rate}.'''				)
										else:
													logger.warning(
													    "It is strongly recommended to pass the `sampling_rate` argument to this function. "
													    "Failing to do so can result in silent errors that might be hard to debug."				)

										snake_case_				:					Optional[int]   =       isinstance(lowerCAmelCase__       ,							np.ndarray				) and len(raw_speech.shape				) > 1
										if is_batched_numpy and len(raw_speech.shape				) > 2:
													raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''				)
										snake_case_				:					Union[str, Any]   =       is_batched_numpy or (
										    isinstance(lowerCAmelCase__       ,							(list, tuple)				) and (isinstance(raw_speech[0]       ,							(np.ndarray, tuple, list)				))
										)

										if is_batched:
													snake_case_				:					List[Any]   =       [np.asarray(lowerCAmelCase__       ,							dtype=np.floataa				) for speech in raw_speech]
										elif not is_batched and not isinstance(lowerCAmelCase__       ,							np.ndarray				):
													snake_case_				:					Optional[int]   =       np.asarray(lowerCAmelCase__       ,							dtype=np.floataa				)
										elif isinstance(lowerCAmelCase__       ,							np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
													snake_case_				:					Optional[int]   =       raw_speech.astype(np.floataa				)

										# always return batch
										if not is_batched:
													snake_case_				:					Tuple   =       [raw_speech]

										# extract fbank features
										snake_case_				:					int   =       [self._extract_fbank_features(lowerCAmelCase__				) for waveform in raw_speech]

										# convert into correct format for padding
										snake_case_				:					int   =       BatchFeature({"input_features": features}				)

										snake_case_				:					Optional[int]   =       self.pad(
										    lowerCAmelCase__       ,							padding=lowerCAmelCase__       ,							max_length=lowerCAmelCase__       ,							truncation=lowerCAmelCase__       ,							pad_to_multiple_of=lowerCAmelCase__       ,							return_attention_mask=lowerCAmelCase__       ,							**lowerCAmelCase__       ,							)

										# make sure list is in array format
										snake_case_				:					str   =       padded_inputs.get("input_features"				)
										if isinstance(input_features[0]       ,							lowerCAmelCase__				):
													snake_case_				:					Tuple   =       [np.asarray(lowerCAmelCase__       ,							dtype=np.floataa				) for feature in input_features]

										snake_case_				:					Any   =       padded_inputs.get("attention_mask"				)
										if attention_mask is not None:
													snake_case_				:					str   =       [np.asarray(lowerCAmelCase__       ,							dtype=np.intaa				) for array in attention_mask]

										# Utterance-level cepstral mean and variance normalization
										if self.do_ceptral_normalize:
													snake_case_				:					int   =       (
													    np.array(lowerCAmelCase__       ,							dtype=np.intaa				)
													    if self._get_padding_strategies(lowerCAmelCase__       ,							max_length=lowerCAmelCase__				) is not PaddingStrategy.DO_NOT_PAD
													    else None
													)
													snake_case_				:					Dict   =       self.normalize(
													    padded_inputs["input_features"]       ,							attention_mask=lowerCAmelCase__				)

										if return_tensors is not None:
													snake_case_				:					List[str]   =       padded_inputs.convert_to_tensors(lowerCAmelCase__				)

										return padded_inputs












'''simple docstring'''




import gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


__lowerCamelCase   :					List[str]   							=					re.compile(R'''\s+''')

def 	__UpperCAmelCase       ( __magic_name__						)->    Union[str, Any]:



			"""simple docstring"""



			return {"hash": hashlib.mda(re.sub(__magic_name__     ,""     ,example["content"]						).encode("utf-8"						)						).hexdigest()}

def 	__UpperCAmelCase       ( __magic_name__						)->    str:



			"""simple docstring"""



			snake_case_				:					Optional[Any]   =       [len(__magic_name__						) for line in example["content"].splitlines()]
			return {"line_mean": np.mean(__magic_name__						), "line_max": max(__magic_name__						)}

def 	__UpperCAmelCase       ( __magic_name__						)->    int:



			"""simple docstring"""



			snake_case_				:					Optional[int]   =       np.mean([c.isalnum() for c in example["content"]]						)
			return {"alpha_frac": alpha_frac}

def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__						)->    Tuple:



			"""simple docstring"""



			if example["hash"] in uniques:
						uniques.remove(example["hash"]						)
						return True
			else:
						return False

def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__=5						)->    Tuple:



			"""simple docstring"""



			snake_case_				:					List[str]   =       ["auto-generated", "autogenerated", "automatically generated"]
			snake_case_				:					Optional[Any]   =       example["content"].splitlines()
			for _, line in zip(range(__magic_name__						)     ,__magic_name__						):
						for keyword in keywords:
									if keyword in line.lower():
												return {"autogenerated": True}
			else:
						return {"autogenerated": False}

def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__=5     ,__magic_name__=0.05						)->    Optional[Any]:



			"""simple docstring"""



			snake_case_				:					str   =       ["unit tests", "test file", "configuration file"]
			snake_case_				:					int   =       example["content"].splitlines()
			snake_case_				:					Optional[Any]   =       0
			snake_case_				:					Any   =       0
			# first test
			for _, line in zip(range(__magic_name__						)     ,__magic_name__						):
						for keyword in keywords:
									if keyword in line.lower():
												return {"config_or_test": True}
    # second test
			snake_case_				:					Tuple   =       example["content"].count("\n"						)
			snake_case_				:					int   =       int(coeff * nlines						)
			for line in lines:
						count_config += line.lower().count("config"						)
						count_test += line.lower().count("test"						)
						if count_config > threshold or count_test > threshold:
									return {"config_or_test": True}
			return {"config_or_test": False}

def 	__UpperCAmelCase       ( __magic_name__						)->    str:



			"""simple docstring"""



			snake_case_				:					List[Any]   =       ["def ", "class ", "for ", "while "]
			snake_case_				:					Optional[Any]   =       example["content"].splitlines()
			for line in lines:
						for keyword in keywords:
									if keyword in line.lower():
												return {"has_no_keywords": False}
			return {"has_no_keywords": True}

def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__=4						)->    Optional[int]:



			"""simple docstring"""



			snake_case_				:					Tuple   =       example["content"].splitlines()
			snake_case_				:					Tuple   =       0
			for line in lines:
						counter += line.lower().count("="						)
						if counter > minimum:
									return {"has_few_assignments": False}
			return {"has_few_assignments": True}

def 	__UpperCAmelCase       ( __magic_name__						)->    List[Any]:



			"""simple docstring"""



			snake_case_				:					Tuple   =       tokenizer(example["content"]     ,truncation=__magic_name__						)["input_ids"]
			snake_case_				:					int   =       len(example["content"]						) / len(__magic_name__						)
			return {"ratio": ratio}

def 	__UpperCAmelCase       ( __magic_name__						)->    Optional[Any]:



			"""simple docstring"""



			snake_case_				:					Union[str, Any]   =       {}
			results.update(get_hash(__magic_name__						)						)
			results.update(line_stats(__magic_name__						)						)
			results.update(alpha_stats(__magic_name__						)						)
			results.update(char_token_ratio(__magic_name__						)						)
			results.update(is_autogenerated(__magic_name__						)						)
			results.update(is_config_or_test(__magic_name__						)						)
			results.update(has_no_keywords(__magic_name__						)						)
			results.update(has_few_assignments(__magic_name__						)						)
			return results

def 	__UpperCAmelCase       ( __magic_name__     ,__magic_name__     ,__magic_name__						)->    Tuple:



			"""simple docstring"""



			if not check_uniques(__magic_name__     ,__magic_name__						):
						return False
			elif example["autogenerated"]:
						return False
			elif example["line_max"] > args.line_max:
						return False
			elif example["line_mean"] > args.line_mean:
						return False
			elif example["alpha_frac"] < args.alpha_frac:
						return False
			elif example["ratio"] < args.min_token_ratio:
						return False
			elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
						return False
			elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
						return False
			elif example["has_few_assignments"]:
						return False
			else:
						return True

def 	__UpperCAmelCase       ( __magic_name__						)->    Dict:



			"""simple docstring"""



			with open(__magic_name__     ,"rb"						) as f_in:
						with gzip.open(str(__magic_name__						) + ".gz"     ,"wb"     ,compresslevel=6						) as f_out:
									shutil.copyfileobj(__magic_name__     ,__magic_name__						)
			os.unlink(__magic_name__						)


# Settings
__lowerCamelCase   :					List[Any]   							=					HfArgumentParser(PreprocessingArguments)
__lowerCamelCase   :					str   							=					parser.parse_args()
if args.num_workers is None:
		__lowerCamelCase   :					List[Any]   							=					multiprocessing.cpu_count()
__lowerCamelCase   :					Optional[int]   							=					AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
__lowerCamelCase   :					Any   							=					time.time()
__lowerCamelCase   :					str   							=					load_dataset(args.dataset_name, split='''train''')
print(f'''Time to load dataset: {time.time()-t_start:.2f}''')

# Run preprocessing
__lowerCamelCase   :					List[str]   							=					time.time()
__lowerCamelCase   :					Any   							=					ds.map(preprocess, num_proc=args.num_workers)
print(f'''Time to preprocess dataset: {time.time()-t_start:.2f}''')

# Deduplicate hashes
__lowerCamelCase   :					Any   							=					set(ds.unique('''hash'''))
__lowerCamelCase   :					Optional[int]   							=					len(uniques) / len(ds)
print(f'''Fraction of duplicates: {1-frac:.2%}''')

# Deduplicate data and apply heuristics
__lowerCamelCase   :					List[str]   							=					time.time()
__lowerCamelCase   :					Tuple   							=					ds.filter(filter, fn_kwargs={'''uniques''': uniques, '''args''': args})
print(f'''Time to filter dataset: {time.time()-t_start:.2f}''')
print(f'''Size of filtered dataset: {len(ds_filter)}''')

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
		__lowerCamelCase   :					List[str]   							=					time.time()
		__lowerCamelCase							,      __lowerCamelCase   :					Tuple   							=					deduplicate_dataset(ds_filter, args.jaccard_threshold)
		print(f'''Time to deduplicate dataset: {time.time()-t_start:.2f}''')
		print(f'''Size of deduplicate dataset: {len(ds_filter)}''')

# Save data in batches of samples_per_file
__lowerCamelCase   :					List[Any]   							=					Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
		with open(output_dir / '''duplicate_clusters.json''', '''w''') as f:
				json.dump(duplicate_clusters, f)

__lowerCamelCase   :					List[str]   							=					output_dir / '''data'''
data_dir.mkdir(exist_ok=True)

__lowerCamelCase   :					int   							=					time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
		__lowerCamelCase   :					Union[str, Any]   							=					str(data_dir / f'''file-{file_number+1:012}.json''')
		__lowerCamelCase   :					List[Any]   							=					min(len(ds_filter), index + args.samples_per_file)
		ds_filter.select(list(range(index, end_index))).to_json(file_path)
		compress_file(file_path)






print(f'''Time to save dataset: {time.time()-t_start:.2f}''')




