from collections import Counter from pathlib import Path import warnings import librosa import numpy as np SAMPLE_RATE = 22050 LABEL_PD = "PD" LABEL_TPD = "TPD" VALID_LABELS = {LABEL_PD, LABEL_TPD} N_MFCC = 13 MIN_DURATION_SECONDS = 2.00 MIN_RMS_FOR_USABLE_AUDIO = 0.001 QUIET_RMS_THRESHOLD = 0.003 CLIPPING_AMPLITUDE_THRESHOLD = 0.99 CLIPPING_RATIO_THRESHOLD = 0.01 FRAME_LENGTH = 2048 HOP_LENGTH = 512 FEATURE_NAMES = ( [f"mfcc_{index}_mean" for index in range(1, N_MFCC + 1)] + [f"mfcc_{index}_std" for index in range(1, N_MFCC + 1)] + [f"delta_mfcc_{index}_mean" for index in range(1, N_MFCC + 1)] + [f"delta_mfcc_{index}_std" for index in range(1, N_MFCC + 1)] + ["rms_normalized_mean", "rms_normalized_std"] + ["rms_mean", "rms_std", "peak_amplitude", "clipping_ratio", "energy_stability"] + ["pitch_mean", "pitch_std", "pitch_range", "pitch_stability", "pitch_variation"] + ["zcr_mean", "zcr_std"] + ["spectral_centroid_mean", "spectral_centroid_std"] + ["spectral_bandwidth_mean", "spectral_bandwidth_std"] + ["spectral_rolloff_mean", "spectral_rolloff_std"] + [ "active_duration", "silence_duration", "silence_ratio", "number_of_pauses", "average_pause_duration", "speech_activity_ratio", ] ) def get_label_from_filename(file_path): """ Mengambil label dari nama file atau nama folder. Urutan pengecekan penting: - cek "_tpd" lebih dulu - baru cek "_pd" """ path = Path(file_path) filename = path.stem.lower() parent = path.parent.name.lower() if "_tpd" in filename or parent in {"tpd", "not_confident", "tidak_percaya_diri"}: return LABEL_TPD if "_pd" in filename or parent in {"pd", "confident", "percaya_diri"}: return LABEL_PD return None def validate_label(label, file_path): if label not in VALID_LABELS: raise ValueError(f"Label tidak valid pada {Path(file_path).name}: {label}") def load_audio_raw(file_path, sample_rate=SAMPLE_RATE): y, sr = librosa.load(file_path, sr=sample_rate, mono=True) if y.size == 0: raise ValueError(f"Audio kosong: {file_path}") return y.astype(np.float32), sr def calculate_audio_quality(y, sr): duration = float(librosa.get_duration(y=y, sr=sr)) if y.size else 0.0 rms_frames = librosa.feature.rms(y=y, frame_length=FRAME_LENGTH, hop_length=HOP_LENGTH)[0] if y.size else np.array([0.0]) rms_mean = float(np.mean(rms_frames)) rms_std = float(np.std(rms_frames)) peak_amplitude = float(np.max(np.abs(y))) if y.size else 0.0 clipping_ratio = float(np.mean(np.abs(y) >= CLIPPING_AMPLITUDE_THRESHOLD)) if y.size else 0.0 return { "duration": duration, "rms_mean": rms_mean, "rms_std": rms_std, "peak_amplitude": peak_amplitude, "clipping_ratio": clipping_ratio, "is_clipped": bool(clipping_ratio > CLIPPING_RATIO_THRESHOLD or peak_amplitude >= CLIPPING_AMPLITUDE_THRESHOLD), "is_too_quiet": bool(rms_mean < QUIET_RMS_THRESHOLD), "is_too_short": bool(duration < MIN_DURATION_SECONDS), } def validate_audio_for_prediction(audio_quality): if audio_quality["is_too_short"]: raise ValueError("Audio terlalu pendek, silakan rekam ulang.") if audio_quality["is_clipped"]: raise ValueError("Audio terlalu keras/pecah, silakan rekam ulang dengan volume normal.") def normalize_audio(y, target_peak=0.95): peak_amplitude = np.max(np.abs(y)) if y.size else 0.0 if peak_amplitude <= 0: return y.astype(np.float32) return (y / peak_amplitude * target_peak).astype(np.float32) def load_and_preprocess_audio(file_path, sample_rate=SAMPLE_RATE, validate_quality=False): """ Membaca audio dengan preprocessing konsisten untuk training dan prediksi: mono, sample rate 22050 Hz, trim silence, dan normalisasi volume. """ y, sr = load_audio_raw(file_path, sample_rate=sample_rate) audio_quality = calculate_audio_quality(y, sr) if validate_quality: validate_audio_for_prediction(audio_quality) y, _ = librosa.effects.trim(y, top_db=30) if y.size == 0: raise ValueError(f"Audio hanya berisi silence: {file_path}") duration = librosa.get_duration(y=y, sr=sr) if validate_quality and duration < MIN_DURATION_SECONDS: raise ValueError( f"Audio terlalu pendek: {duration:.2f} detik. Minimal {MIN_DURATION_SECONDS:.2f} detik." ) rms_value = float(np.sqrt(np.mean(y**2))) if rms_value < MIN_RMS_FOR_USABLE_AUDIO: raise ValueError( f"Audio terlalu pelan/silent. RMS={rms_value:.5f}, " f"minimal {MIN_RMS_FOR_USABLE_AUDIO:.5f}." ) y = normalize_audio(y) return y.astype(np.float32), sr def mean_std(feature_matrix): """ Mengubah fitur frame-based menjadi statistik tetap. Output selalu 1 dimensi dan stabil untuk SVM. """ feature_matrix = np.atleast_2d(feature_matrix) return np.concatenate( [ np.mean(feature_matrix, axis=1), np.std(feature_matrix, axis=1), ] ) def extract_pitch_features(y, sr): """ Mengambil ringkasan fundamental frequency (pitch) dengan pyin. Jika pitch tidak terdeteksi, nilai pitch dibuat 0 agar fitur tetap konsisten. """ f0, _, _ = librosa.pyin( y, fmin=librosa.note_to_hz("C2"), fmax=librosa.note_to_hz("C7"), sr=sr, ) voiced_f0 = f0[~np.isnan(f0)] if voiced_f0.size == 0: return { "pitch_mean": 0.0, "pitch_std": 0.0, "pitch_range": 0.0, "pitch_stability": 0.0, "pitch_variation": 0.0, } pitch_mean = float(np.mean(voiced_f0)) pitch_std = float(np.std(voiced_f0)) pitch_range = float(np.max(voiced_f0) - np.min(voiced_f0)) pitch_variation = float(pitch_std / (pitch_mean + 1e-8)) pitch_stability = float(1.0 / (1.0 + pitch_variation)) return { "pitch_mean": pitch_mean, "pitch_std": pitch_std, "pitch_range": pitch_range, "pitch_stability": pitch_stability, "pitch_variation": pitch_variation, } def extract_pause_features(y, sr): y_trimmed, _ = librosa.effects.trim(y, top_db=30) if y_trimmed.size == 0: y_trimmed = y total_duration = float(librosa.get_duration(y=y_trimmed, sr=sr)) rms_frames = librosa.feature.rms(y=y_trimmed, frame_length=FRAME_LENGTH, hop_length=HOP_LENGTH)[0] if rms_frames.size == 0 or total_duration <= 0: return { "active_duration": 0.0, "silence_duration": total_duration, "silence_ratio": 1.0, "number_of_pauses": 0.0, "average_pause_duration": 0.0, "speech_activity_ratio": 0.0, } max_rms = float(np.max(rms_frames)) median_rms = float(np.median(rms_frames)) adaptive_threshold = max(0.0006, max_rms * 0.06, median_rms * 0.55) active_frames = rms_frames > adaptive_threshold frame_duration = HOP_LENGTH / sr # Lubang diam yang sangat pendek masih dianggap bagian dari artikulasi normal. min_gap_frames = int(np.ceil(0.18 / frame_duration)) smoothed_active_frames = active_frames.copy() inactive_run = 0 inactive_start = 0 for index, is_active in enumerate(active_frames): if not is_active: if inactive_run == 0: inactive_start = index inactive_run += 1 else: if 0 < inactive_run < min_gap_frames: smoothed_active_frames[inactive_start:index] = True inactive_run = 0 if 0 < inactive_run < min_gap_frames: smoothed_active_frames[inactive_start:] = True active_duration = float(min(total_duration, np.sum(smoothed_active_frames) * frame_duration)) silence_duration = float(max(0.0, total_duration - active_duration)) silence_ratio = float(silence_duration / (total_duration + 1e-8)) speech_activity_ratio = float(active_duration / (total_duration + 1e-8)) pauses = [] inactive_run = 0 for is_active in smoothed_active_frames: if is_active: if inactive_run * frame_duration >= 0.30: pauses.append(inactive_run * frame_duration) inactive_run = 0 else: inactive_run += 1 if inactive_run * frame_duration >= 0.30: pauses.append(inactive_run * frame_duration) return { "active_duration": active_duration, "silence_duration": silence_duration, "silence_ratio": silence_ratio, "number_of_pauses": float(len(pauses)), "average_pause_duration": float(np.mean(pauses)) if pauses else 0.0, "speech_activity_ratio": speech_activity_ratio, } def calculate_voice_indicator_scores(raw_quality, pitch_features, pause_features): rms_mean = raw_quality["rms_mean"] rms_std = raw_quality["rms_std"] clipping_ratio = raw_quality["clipping_ratio"] peak_amplitude = raw_quality["peak_amplitude"] volume_strength = np.clip( np.log10((rms_mean + 1e-8) / QUIET_RMS_THRESHOLD + 1.0) / np.log10(14.0), 0.0, 1.0, ) energy_stability = 1.0 / (1.0 + (rms_std / (rms_mean + 1e-8))) clipping_penalty = np.clip((clipping_ratio / CLIPPING_RATIO_THRESHOLD) + max(0.0, peak_amplitude - 0.95) * 10.0, 0.0, 1.0) volume_score = float(np.clip((0.65 * volume_strength + 0.35 * energy_stability) * (1.0 - clipping_penalty), 0.0, 1.0)) pitch_variation = pitch_features["pitch_variation"] pitch_range = pitch_features["pitch_range"] natural_variation = np.clip(pitch_range / 180.0, 0.0, 1.0) not_flat = np.clip(pitch_variation / 0.08, 0.0, 1.0) not_shaky = 1.0 - np.clip(max(0.0, pitch_variation - 0.35) / 0.45, 0.0, 1.0) intonation_score = float(np.clip(0.35 * natural_variation + 0.30 * not_flat + 0.35 * not_shaky, 0.0, 1.0)) silence_ratio = pause_features["silence_ratio"] average_pause_duration = pause_features["average_pause_duration"] number_of_pauses = pause_features["number_of_pauses"] silence_penalty = np.clip((silence_ratio - 0.12) / 0.38, 0.0, 1.0) pause_count_penalty = np.clip(number_of_pauses / 5.0, 0.0, 1.0) pause_duration_penalty = np.clip(average_pause_duration / 0.90, 0.0, 1.0) pause_score = 1.0 - (0.55 * silence_penalty + 0.25 * pause_count_penalty + 0.20 * pause_duration_penalty) return { "volume_score": float(np.clip(volume_score, 0.0, 1.0)), "intonation_score": float(np.clip(intonation_score, 0.0, 1.0)), "pause_score": float(np.clip(pause_score, 0.0, 1.0)), "speech_activity_ratio": float(np.clip(pause_features["speech_activity_ratio"], 0.0, 1.0)), "silence_ratio": float(np.clip(silence_ratio, 0.0, 1.0)), } def build_prediction_explanation(predicted_label, confidence, indicators): weak_points = [] strong_points = [] if indicators["volume_score"] >= 0.65: strong_points.append("volume stabil") else: weak_points.append("volume kurang stabil atau kurang jelas") if indicators["intonation_score"] >= 0.65: strong_points.append("intonasi cukup bervariasi") else: weak_points.append("intonasi kurang stabil") if indicators["pause_score"] >= 0.65: strong_points.append("jeda bicara sedikit dan wajar") else: weak_points.append("jeda bicara cukup banyak") if confidence < 0.60: reason = " dan ".join(weak_points[:2]) if weak_points else "indikator suara saling berdekatan" return f"Model belum cukup yakin karena {reason}." if predicted_label == LABEL_PD: reason = ", ".join(strong_points[:2]) if len(strong_points) > 2: reason += f", dan {strong_points[2]}" return f"Suara terdeteksi percaya diri karena {reason}." reason = " dan ".join(weak_points[:2]) if weak_points else "kombinasi indikator belum menunjukkan kestabilan yang cukup" return f"Suara terdeteksi tidak percaya diri karena {reason}." def analyze_audio(file_path, sample_rate=SAMPLE_RATE, validate_quality=False): raw_y, sr = load_audio_raw(file_path, sample_rate=sample_rate) raw_quality = calculate_audio_quality(raw_y, sr) if validate_quality: validate_audio_for_prediction(raw_quality) y, sr = load_and_preprocess_audio(file_path, sample_rate=sample_rate, validate_quality=False) mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=N_MFCC, hop_length=HOP_LENGTH) delta_mfcc = librosa.feature.delta(mfcc) rms_normalized = librosa.feature.rms(y=y, frame_length=FRAME_LENGTH, hop_length=HOP_LENGTH) rms_raw = librosa.feature.rms(y=raw_y, frame_length=FRAME_LENGTH, hop_length=HOP_LENGTH)[0] zcr = librosa.feature.zero_crossing_rate(y, frame_length=FRAME_LENGTH, hop_length=HOP_LENGTH) spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr, hop_length=HOP_LENGTH) spectral_bandwidth = librosa.feature.spectral_bandwidth(y=y, sr=sr, hop_length=HOP_LENGTH) spectral_rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr, hop_length=HOP_LENGTH) pitch_features = extract_pitch_features(y, sr) pause_features = extract_pause_features(raw_y, sr) rms_mean = float(np.mean(rms_raw)) rms_std = float(np.std(rms_raw)) energy_stability = float(1.0 / (1.0 + (rms_std / (rms_mean + 1e-8)))) scalar_features = { "rms_mean": rms_mean, "rms_std": rms_std, "peak_amplitude": raw_quality["peak_amplitude"], "clipping_ratio": raw_quality["clipping_ratio"], "energy_stability": energy_stability, **pitch_features, **pause_features, } feature_vector = np.concatenate( [ mean_std(mfcc), mean_std(delta_mfcc), mean_std(rms_normalized), np.array( [ scalar_features["rms_mean"], scalar_features["rms_std"], scalar_features["peak_amplitude"], scalar_features["clipping_ratio"], scalar_features["energy_stability"], scalar_features["pitch_mean"], scalar_features["pitch_std"], scalar_features["pitch_range"], scalar_features["pitch_stability"], scalar_features["pitch_variation"], ], dtype=np.float32, ), mean_std(zcr), mean_std(spectral_centroid), mean_std(spectral_bandwidth), mean_std(spectral_rolloff), np.array( [ scalar_features["active_duration"], scalar_features["silence_duration"], scalar_features["silence_ratio"], scalar_features["number_of_pauses"], scalar_features["average_pause_duration"], scalar_features["speech_activity_ratio"], ], dtype=np.float32, ), ] ) if feature_vector.ndim != 1: raise ValueError("Fitur audio harus 1 dimensi.") if feature_vector.size != len(FEATURE_NAMES): raise ValueError( f"Jumlah fitur tidak konsisten: {feature_vector.size}, seharusnya {len(FEATURE_NAMES)}." ) if not np.all(np.isfinite(feature_vector)): raise ValueError("Fitur audio mengandung NaN atau infinity.") indicators = calculate_voice_indicator_scores(raw_quality, pitch_features, pause_features) return { "features": feature_vector.astype(np.float32), "audio_quality": raw_quality, "voice_indicators": indicators, "feature_details": scalar_features, } def extract_features(file_path, sample_rate=SAMPLE_RATE): """ Ekstraksi fitur suara yang sama untuk training dan prediksi: - MFCC mean dan std - RMS Energy mean dan std - Zero Crossing Rate mean dan std - Spectral Centroid mean dan std - Spectral Bandwidth mean dan std - Spectral Rolloff mean dan std - Pitch/fundamental frequency - Durasi suara aktif """ return analyze_audio(file_path, sample_rate=sample_rate)["features"] def load_dataset(data_dir): """ Membaca semua file .wav pada folder data. File tanpa label valid atau file rusak dilewati dengan peringatan. """ data_path = Path(data_dir) audio_files = sorted(data_path.rglob("*.wav")) if not audio_files: raise FileNotFoundError(f"Tidak ada file .wav di folder: {data_path}") features = [] labels = [] used_files = [] for audio_file in audio_files: label = get_label_from_filename(audio_file) if label is None: warnings.warn( f"File dilewati karena nama/folder tidak mengandung label PD atau TPD: " f"{audio_file.name}" ) continue try: validate_label(label, audio_file) features.append(extract_features(audio_file)) labels.append(label) used_files.append(audio_file) except Exception as error: warnings.warn(f"File dilewati karena gagal diproses: {audio_file.name} ({error})") if not features: raise ValueError("Tidak ada file audio valid yang berhasil diproses.") label_counts = Counter(labels) print("\n=== Distribusi Label Dataset ===") print(f"PD : {label_counts.get(LABEL_PD, 0)}") print(f"TPD: {label_counts.get(LABEL_TPD, 0)}") invalid_labels = set(labels) - VALID_LABELS if invalid_labels: raise ValueError(f"Ditemukan label tidak valid: {sorted(invalid_labels)}") return np.array(features), np.array(labels), used_files def check_dataset_quality(data_dir): """ Mengecek kualitas dataset: - jumlah data PD dan TPD - durasi setiap audio - audio terlalu pendek - audio terlalu pelan/silent - file rusak - rekomendasi file yang perlu direkam ulang """ data_path = Path(data_dir) audio_files = sorted(data_path.rglob("*.wav")) label_counts = Counter() problems = [] print("\n=== Cek Kualitas Dataset ===") for audio_file in audio_files: label = get_label_from_filename(audio_file) if label is None: problems.append((audio_file.name, "Label tidak ditemukan")) continue label_counts[label] += 1 try: y_raw, sr = librosa.load(audio_file, sr=SAMPLE_RATE, mono=True) duration_raw = librosa.get_duration(y=y_raw, sr=sr) rms_raw = float(np.sqrt(np.mean(y_raw**2))) if y_raw.size else 0.0 issue_notes = [] if duration_raw < MIN_DURATION_SECONDS: issue_notes.append(f"terlalu pendek ({duration_raw:.2f} detik)") if rms_raw < MIN_RMS_FOR_USABLE_AUDIO: issue_notes.append(f"terlalu pelan/silent (RMS={rms_raw:.5f})") print( f"{audio_file.name} | label={label} | durasi={duration_raw:.2f}s | " f"rms={rms_raw:.5f}" ) if issue_notes: problems.append((audio_file.name, ", ".join(issue_notes))) except Exception as error: problems.append((audio_file.name, f"file rusak/gagal dibaca ({error})")) print("\nJumlah data:") print(f"PD : {label_counts.get(LABEL_PD, 0)}") print(f"TPD: {label_counts.get(LABEL_TPD, 0)}") print("\nRekomendasi rekam ulang/perbaikan:") if not problems: print("Tidak ada masalah kualitas audio yang jelas.") else: for filename, reason in problems: print(f"- {filename}: {reason}") return problems