import os import tempfile import warnings from pathlib import Path import joblib import librosa import noisereduce as nr import numpy as np import opensmile import soundfile as sf import xgboost as xgb from sklearn.metrics import ( accuracy_score, confusion_matrix, f1_score, precision_score, recall_score, roc_auc_score, ) warnings.filterwarnings("ignore") # Audio configuration must match training pipeline. TARGET_SR = 16000 MIN_DURATION_S = 1.5 MAX_DURATION_S = 12.0 TOP_DB = 30 NOISE_REDUCE = True smile = opensmile.Smile( feature_set=opensmile.FeatureSet.eGeMAPSv02, feature_level=opensmile.FeatureLevel.Functionals, ) def load_and_preprocess(file_path, augment=False): """Load and preprocess audio with the same steps used in training.""" try: audio, sr = librosa.load(file_path, sr=None, mono=True) if sr != TARGET_SR: audio = librosa.resample(audio, orig_sr=sr, target_sr=TARGET_SR) peak = np.max(np.abs(audio)) if peak > 0: audio = audio / peak audio, _ = librosa.effects.trim(audio, top_db=TOP_DB) duration = len(audio) / TARGET_SR if duration < MIN_DURATION_S: return None if duration > MAX_DURATION_S: max_samples = int(MAX_DURATION_S * TARGET_SR) start = (len(audio) - max_samples) // 2 audio = audio[start : start + max_samples] if NOISE_REDUCE: noise_sample_len = min(int(0.3 * TARGET_SR), len(audio)) if noise_sample_len > 0: noise_profile = audio[:noise_sample_len] audio = nr.reduce_noise( y=audio, sr=TARGET_SR, y_noise=noise_profile, prop_decrease=0.75, stationary=False, ) # Kept for API compatibility with the notebook pipeline. _ = augment return audio except Exception: return None def extract_features(audio_array, sr=TARGET_SR): """Extract eGeMAPSv02 features from a preprocessed audio numpy array.""" tmp_path = None try: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: tmp_path = tmp.name sf.write(tmp_path, audio_array, sr, subtype="PCM_16") df = smile.process_file(tmp_path) return df.values.flatten().astype(np.float32) except Exception: return None finally: if tmp_path and os.path.exists(tmp_path): os.unlink(tmp_path) class VoicePDPredictor: def __init__( self, model_path="opxg_model.json", scaler_path="scaler.pkl", threshold_path="threshold.txt", feature_names_path="feature_names.csv", ): self.model = xgb.XGBClassifier() self.model.load_model(model_path) self.scaler = joblib.load(scaler_path) with open(threshold_path, "r", encoding="utf-8") as f: self.threshold = float(f.read().strip()) self.expected_feature_count = sum(1 for _ in open(feature_names_path, "r", encoding="utf-8")) print(f"VoicePDPredictor loaded | threshold={self.threshold:.4f}") def predict(self, file_path): audio = load_and_preprocess(file_path, augment=False) if audio is None: return {"error": "preprocessing_failed"} features = extract_features(audio) if features is None: return {"error": "feature_extraction_failed"} if features.shape[0] != self.expected_feature_count: return {"error": f"feature_mismatch_{features.shape[0]}"} features = np.nan_to_num(features, nan=0.0).reshape(1, -1) features_scaled = self.scaler.transform(features) prob = float(self.model.predict_proba(features_scaled)[0][1]) label = "Parkinsons" if prob >= self.threshold else "Healthy" return {"label": label, "probability": prob, "error": None} def _collect_audio_files(folder): exts = {".wav", ".mp3", ".ogg", ".flac", ".m4a"} return sorted([p for p in Path(folder).rglob("*") if p.suffix.lower() in exts]) def evaluate_test_folders(predictor, base_dir="test_audios"): healthy_files = _collect_audio_files(Path(base_dir) / "healthy") pd_files = _collect_audio_files(Path(base_dir) / "pd") y_true = [] y_pred = [] y_prob = [] failed = [] for file_path in healthy_files: result = predictor.predict(str(file_path)) if result["error"]: failed.append((str(file_path), result["error"])) continue y_true.append(0) y_pred.append(1 if result["label"] == "Parkinsons" else 0) y_prob.append(result["probability"]) for file_path in pd_files: result = predictor.predict(str(file_path)) if result["error"]: failed.append((str(file_path), result["error"])) continue y_true.append(1) y_pred.append(1 if result["label"] == "Parkinsons" else 0) y_prob.append(result["probability"]) if not y_true: raise RuntimeError("No valid audio files were processed.") tn, fp, fn, tp = confusion_matrix(y_true, y_pred, labels=[0, 1]).ravel() metrics = { "healthy_total": len(healthy_files), "pd_total": len(pd_files), "used_samples": len(y_true), "failed_samples": len(failed), "accuracy": accuracy_score(y_true, y_pred), "precision_pd": precision_score(y_true, y_pred, zero_division=0), "recall_pd": recall_score(y_true, y_pred, zero_division=0), "f1_pd": f1_score(y_true, y_pred, zero_division=0), "specificity_healthy": (tn / (tn + fp)) if (tn + fp) > 0 else 0.0, "roc_auc": roc_auc_score(y_true, y_prob) if len(set(y_true)) > 1 else float("nan"), "confusion_matrix": {"tn": int(tn), "fp": int(fp), "fn": int(fn), "tp": int(tp)}, "failed_examples": failed[:10], } return metrics if __name__ == "__main__": predictor = VoicePDPredictor( model_path="opxg_model.json", scaler_path="scaler.pkl", threshold_path="threshold.txt", feature_names_path="feature_names.csv", ) metrics = evaluate_test_folders(predictor, base_dir="test_audios") print("\n=== Inference Metrics on test_audios ===") print(f"Healthy files : {metrics['healthy_total']}") print(f"PD files : {metrics['pd_total']}") print(f"Used samples : {metrics['used_samples']}") print(f"Failed samples : {metrics['failed_samples']}") print(f"Accuracy : {metrics['accuracy']:.4f}") print(f"Precision (PD) : {metrics['precision_pd']:.4f}") print(f"Recall (PD) : {metrics['recall_pd']:.4f}") print(f"F1-score (PD) : {metrics['f1_pd']:.4f}") print(f"Specificity (Healthy): {metrics['specificity_healthy']:.4f}") print(f"ROC-AUC : {metrics['roc_auc']:.4f}") cm = metrics["confusion_matrix"] print( "Confusion Matrix [Healthy=0, PD=1] -> " f"TN={cm['tn']} FP={cm['fp']} FN={cm['fn']} TP={cm['tp']}" ) if metrics["failed_examples"]: print("\nFailed examples (up to 10):") for path, error in metrics["failed_examples"]: print(f" {path} -> {error}")