Spaces:
Sleeping
Sleeping
File size: 3,925 Bytes
dead0b1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import os
import pandas as pd
import librosa
import soundfile as sf
import numpy as np
from tqdm import tqdm
import sys
# Add src to path
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from src.config import RAW_HUMAN_DIR, RAW_AI_DIR, PROCESSED_DIR, SAMPLE_RATE, DURATION_LIMIT, DATA_DIR
def preprocess_audio(file_path, output_path):
"""
Standardize audio:
- Load as Mono
- Resample to 16kHz
- Trim silence
- Normalize amplitude
- Pad/Trim to fixed duration (optional, but good for batching, let's just ensure min length for now)
"""
try:
# Load audio (librosa handles resampling and mono conversion)
y, sr = librosa.load(file_path, sr=SAMPLE_RATE, mono=True)
# Trim silence (top_db=20 is a standard threshold)
y_trimmed, _ = librosa.effects.trim(y, top_db=20)
# Skip if too short (less than 0.5s)
if len(y_trimmed) < 0.5 * SAMPLE_RATE:
return False, "Too short"
# Normalize amplitude (Peak normalization)
y_norm = librosa.util.normalize(y_trimmed)
# Save processed file
sf.write(output_path, y_norm, SAMPLE_RATE)
return True, "Success"
except Exception as e:
return False, str(e)
def process_dataset(input_csv, source_type):
"""
Process all files listed in the CSV
source_type: 'human' or 'ai'
"""
if not os.path.exists(input_csv):
print(f"Dataset CSV not found: {input_csv}")
return []
df = pd.read_csv(input_csv)
processed_records = []
output_dir = os.path.join(PROCESSED_DIR, source_type)
if not os.path.exists(output_dir):
os.makedirs(output_dir)
print(f"Processing {source_type} samples...")
for _, row in tqdm(df.iterrows(), total=len(df)):
file_path = row['path']
filename = row['filename']
lang = row['language']
# Create language subdir in processed
lang_dir = os.path.join(output_dir, lang)
if not os.path.exists(lang_dir):
os.makedirs(lang_dir)
output_filename = f"proc_{filename}"
if not output_filename.endswith('.wav'):
# Enforce wav for processed data usually, or keep original extension if flac/mp3 is fine.
# wav is safer for downstream processing.
output_filename = os.path.splitext(output_filename)[0] + ".wav"
output_path = os.path.join(lang_dir, output_filename)
success, msg = preprocess_audio(file_path, output_path)
if success:
processed_records.append({
'filename': output_filename,
'original_filename': filename,
'path': output_path,
'label': source_type, # 'human' or 'ai'
'language': lang,
'split': 'train' # Default, will split later
})
return processed_records
def main():
if not os.path.exists(PROCESSED_DIR):
os.makedirs(PROCESSED_DIR)
all_processed = []
# Process Human Data
human_csv = os.path.join(RAW_HUMAN_DIR, 'human_samples.csv')
human_data = process_dataset(human_csv, 'human')
all_processed.extend(human_data)
# Process AI Data
ai_csv = os.path.join(RAW_AI_DIR, 'ai_samples.csv')
ai_data = process_dataset(ai_csv, 'ai')
all_processed.extend(ai_data)
# Save Master Dataset
master_df = pd.DataFrame(all_processed)
master_csv = os.path.join(DATA_DIR, 'master_dataset.csv')
master_df.to_csv(master_csv, index=False)
# Print Stats
print("\nProcessing Complete!")
print(f"Total Processed Samples: {len(master_df)}")
print(master_df['label'].value_counts())
print(master_df['language'].value_counts())
if __name__ == "__main__":
main()
|