Upload folder using huggingface_hub

app/main.py

@@ -94,7 +94,7 @@ def predict():
         
         try:
             load_resources()
-            y, sr = librosa.load(filepath, sr=16000, duration=5.0)
+            y, sr = librosa.load(filepath, sr=16000)
             
             # VAD Lite
             rms_energy = np.mean(librosa.feature.rms(y=y))

test_real_pipeline.py

@@ -0,0 +1,121 @@
+import os
+import sys
+import numpy as np
+import librosa
+import tensorflow as tf
+from tqdm import tqdm
+import random
+
+# Force TF logs to error only
+os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
+
+PROJECT_ROOT = r"c:\Users\ASUS\lung_ai_project"
+sys.path.insert(0, PROJECT_ROOT)
+
+from utils.hear_extractor import HeARExtractor
+from utils.audio_preprocessor import advanced_preprocess
+
+# Paths
+AUDIO_ROOT = r"c:\Users\ASUS\lung_ai_project\data\coughvid_public\organized"
+SICK_DIR = os.path.join(AUDIO_ROOT, "sick")
+HEALTHY_DIR = os.path.join(AUDIO_ROOT, "healthy")
+MODEL_PATH = os.path.join(PROJECT_ROOT, "models", "hear_classifier_v8_elite.h5")
+
+META_PATH = r"c:\Users\ASUS\lung_ai_project\data\coughvid_public\metadata_compiled.csv"
+import pandas as pd
+
+def get_audio_files(directory, count=50, is_sick=True):
+    df = pd.read_csv(META_PATH)
+    if is_sick:
+        uuids = df[(df['status'].isin(['COVID-19', 'sick'])) & (df['cough_detected'] > 0.8) & (df['SNR'] > 5)]['uuid'].tolist()
+    else:
+        uuids = df[(df['status'] == 'healthy') & (df['cough_detected'] > 0.95)]['uuid'].tolist()
+        
+    random.seed(42)
+    uuids = random.sample(uuids, count * 3) # Sample more in case some missing
+    
+    files = []
+    for u in uuids:
+        for ext in ['.webm', '.wav', '.ogg']:
+            p1 = os.path.join(directory, f"cv_{u}{ext}")
+            p2 = os.path.join(directory, f"{u}{ext}")
+            if os.path.exists(p1):
+                files.append(p1)
+                break
+            if os.path.exists(p2):
+                files.append(p2)
+                break
+        if len(files) == count:
+            break
+    return files
+
+def test_pipeline():
+    print("Loading V8 Elite Model & HeAR Extractor...")
+    classifier_model = tf.keras.models.load_model(MODEL_PATH, compile=False)
+    extractor = HeARExtractor()
+    
+    sick_files = get_audio_files(SICK_DIR, 50, True)
+    healthy_files = get_audio_files(HEALTHY_DIR, 50, False)
+    
+    test_cases = [(f, "sick") for f in sick_files] + [(f, "healthy") for f in healthy_files]
+    
+    correct_healthy = 0
+    correct_sick = 0
+    total_healthy = 0
+    total_sick = 0
+    
+    print("\nStarting Real-World API Pipeline Test (100 Samples)...")
+    for file_path, true_label in tqdm(test_cases):
+        try:
+            # 1. API Load (Full duration!)
+            y, sr = librosa.load(file_path, sr=16000)
+            
+            # 2. API VAD
+            rms_energy = np.mean(librosa.feature.rms(y=y))
+            if rms_energy < 0.005:
+                # API rejects quiet audio. For testing, skip.
+                continue
+                
+            # 3. API Preprocess
+            y_clean = advanced_preprocess(y, sr)
+            
+            # 4. API Extract
+            emb = extractor.extract(y_clean)
+            if emb is None:
+                continue
+                
+            # 5. API Predict
+            X = emb[np.newaxis, ...]
+            prob = classifier_model.predict(X, verbose=0)[0][0]
+            
+            # 6. API Logic
+            THRESHOLD = 0.50
+            if prob > THRESHOLD:
+                final_label = "sick"
+            else:
+                final_label = "healthy"
+                
+            if true_label == "healthy":
+                total_healthy += 1
+                if final_label == "healthy": correct_healthy += 1
+            else:
+                total_sick += 1
+                if final_label == "sick": correct_sick += 1
+                
+        except Exception as e:
+            continue
+            
+    print("\n" + "="*50)
+    print(" FINAL API PIPELINE RESULTS")
+    print("="*50)
+    
+    if total_healthy > 0:
+        h_acc = (correct_healthy / total_healthy) * 100
+        print(f"✅ HEALTHY Accuracy: {h_acc:.2f}% ({correct_healthy}/{total_healthy})")
+    
+    if total_sick > 0:
+        s_acc = (correct_sick / total_sick) * 100
+        print(f"🦠 SICK Accuracy:    {s_acc:.2f}% ({correct_sick}/{total_sick})")
+
+if __name__ == "__main__":
+    test_pipeline()

utils/augment_phone_mic.py

@@ -0,0 +1,65 @@
+import os
+import sys
+import numpy as np
+import librosa
+import soundfile as sf
+import warnings
+from tqdm import tqdm
+
+"""
+PHASE 9: TELEPHONY & WEBM DATA AUGMENTATION
+To prevent healthy patients from being flagged as "High Risk" when recording from 
+cheap laptop mics or browsers (WebM compression artifacts), we must intentionally 
+degrade our training dataset to teach the HeAR AI what a "Phone" sounds like.
+"""
+
+def degrade_audio_to_phone_quality(audio, sr):
+    """
+    Simulates a cheap, compressed smartphone microphone over a web browser.
+    1. Bandpass filter to isolate 300Hz - 3400Hz (Standard Telephony Band)
+    2. Intentional Downsampling & Quantization (Bit-crushing to simulate compression)
+    3. Hiss Injection (Simulating mic hardware noise floor)
+    """
+    import scipy.signal as signal
+    
+    # 1. Telephony Bandpass (300Hz - 3400Hz)
+    nyquist = sr / 2.0
+    low = 300.0 / nyquist
+    high = 3400.0 / nyquist
+    b, a = signal.butter(4, [low, high], btype='band')
+    audio_bandpassed = signal.filtfilt(b, a, audio)
+    
+    # 2. Add Hiss (White Noise Floor)
+    noise_amp = 0.005 * np.random.uniform()
+    white_noise = np.random.randn(len(audio_bandpassed)) * noise_amp
+    audio_noisy = audio_bandpassed + white_noise
+    
+    # 3. 8-bit Quantization (Crushing resolution mimicking bad WebM encoding)
+    audio_noisy = np.clip(audio_noisy, -1.0, 1.0)
+    audio_quantized = np.round(audio_noisy * 128) / 128.0
+    
+    return audio_quantized
+
+def process_directory(source_dir, output_dir):
+    os.makedirs(output_dir, exist_ok=True)
+    files = [f for f in os.listdir(source_dir) if f.endswith('.wav')]
+    
+    for f in tqdm(files, desc=f"Corrupting {os.path.basename(source_dir)}"):
+        try:
+            path = os.path.join(source_dir, f)
+            y, sr = librosa.load(path, sr=16000)
+            y_degraded = degrade_audio_to_phone_quality(y, sr)
+            
+            out_path = os.path.join(output_dir, f"phone_{f}")
+            sf.write(out_path, y_degraded, sr)
+        except:
+            pass
+
+if __name__ == "__main__":
+    base = r"c:\Users\ASUS\lung_ai_project\data\cough"
+    target = r"c:\Users\ASUS\lung_ai_project\data\cough_phone_augmented"
+    
+    # Corrupt both classes
+    process_directory(os.path.join(base, "sick"), os.path.join(target, "sick"))
+    process_directory(os.path.join(base, "healthy"), os.path.join(target, "healthy"))
+    print("Phone Simulation Dataset Created. Ready for V9 Elite Training.")