Create app.py

app.py

@@ -0,0 +1,148 @@
+import os
+import shutil
+import numpy as np
+import librosa
+import noisereduce as nr
+import scipy.signal as signal
+from fastapi import FastAPI, UploadFile, File, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+
+app = FastAPI()
+
+# Enable CORS for Flutter mobile access
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# --- CONFIGURATION ---
+SR = 16000
+SILENCE_DB = 25
+APNEA_MIN = 10.0
+APNEA_MAX = 120.0
+
+# --- CORE LOGIC FUNCTIONS ---
+
+def clean_audio_stream(y, sr):
+    """Clean audio with noise reduction and bandpass filter"""
+    y_denoised = nr.reduce_noise(y=y, sr=sr)
+    b, a = signal.butter(4, [200/(sr/2), 2000/(sr/2)], btype='band')
+    y_filtered = signal.filtfilt(b, a, y_denoised)
+    return y_filtered
+
+def is_snoring_sound(y_segment, sr):
+    """Spectral analysis for snoring detection"""
+    if len(y_segment) < sr * 0.15:
+        return False, 0.0
+    
+    rms_energy = np.sqrt(np.mean(y_segment**2))
+    if rms_energy < 0.008 or rms_energy > 0.95:
+        return False, 0.0
+    
+    stft = np.abs(librosa.stft(y_segment))
+    freq_bins = librosa.fft_frequencies(sr=sr)
+    
+    total_energy = np.sum(stft) + 1e-10
+    snoring_ratio = np.sum(stft[(freq_bins >= 200) & (freq_bins <= 800), :]) / total_energy
+    
+    zcr = np.mean(librosa.feature.zero_crossing_rate(y_segment)[0])
+    centroid = np.mean(librosa.feature.spectral_centroid(y=y_segment, sr=sr)[0])
+    
+    confidence = 0.0
+    if rms_energy >= 0.008: confidence += 0.25
+    confidence += min(snoring_ratio / 0.15, 1.0) * 0.30
+    if centroid <= 2200: confidence += 0.15
+    if zcr <= 0.25: confidence += 0.10
+    
+    return confidence >= 0.40, round(confidence, 2)
+
+def is_human_breathing_sound(y_segment, sr):
+    """Detect quiet breathing"""
+    rms_energy = np.sqrt(np.mean(y_segment**2))
+    if rms_energy < 0.002 or rms_energy > 0.5:
+        return False
+    centroid = np.mean(librosa.feature.spectral_centroid(y=y_segment, sr=sr)[0])
+    return centroid <= 2000
+
+# --- API ROUTES ---
+
+@app.get("/")
+def home():
+    return {"status": "Sleep API is Live", "model": "Signal Processing v2"}
+
+@app.post("/analyze")
+async def analyze_endpoint(file: UploadFile = File(...)):
+    # 1. Save file temporarily
+    temp_path = f"temp_{file.filename}"
+    with open(temp_path, "wb") as buffer:
+        shutil.copyfileobj(file.file, buffer)
+    
+    try:
+        # 2. Load Audio
+        y_orig, sr = librosa.load(temp_path, sr=SR)
+        duration = len(y_orig) / sr
+        
+        # 3. Quick Validation
+        if duration < 5:
+            raise HTTPException(status_code=400, detail="Audio too short for analysis")
+
+        # 4. Process and Detect
+        y_clean = clean_audio_stream(y_orig, sr)
+        intervals = librosa.effects.split(y_clean, top_db=SILENCE_DB)
+        
+        events = []
+        prev_end = 0
+        
+        for idx, (start, end) in enumerate(intervals):
+            # Check for Apnea Gap
+            gap_dur = (start - prev_end) / sr
+            if APNEA_MIN <= gap_dur <= APNEA_MAX:
+                events.append({
+                    "label": "APNEA",
+                    "start": round(prev_end/sr, 2),
+                    "end": round(start/sr, 2),
+                    "duration": round(gap_dur, 2)
+                })
+            
+            # Check Sound for Snoring
+            seg = y_orig[start:end]
+            snore_bool, conf = is_snoring_sound(seg, sr)
+            if snore_bool:
+                events.append({
+                    "label": "SNORING",
+                    "start": round(start/sr, 2),
+                    "end": round(end/sr, 2),
+                    "confidence": conf
+                })
+            prev_end = end
+
+        # 5. Finalize Statistics
+        apnea_list = [e for e in events if e["label"] == "APNEA"]
+        ahi = len(apnea_list) / (duration / 3600) if duration > 0 else 0
+        
+        severity = "NORMAL"
+        if ahi >= 30: severity = "SEVERE"
+        elif ahi >= 15: severity = "MODERATE"
+        elif ahi >= 5: severity = "MILD"
+
+        return {
+            "success": True,
+            "filename": file.filename,
+            "duration": round(duration, 2),
+            "ahi": round(ahi, 2),
+            "severity": severity,
+            "total_events": len(events),
+            "events": events
+        }
+
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+    finally:
+        if os.path.exists(temp_path):
+            os.remove(temp_path)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)