Update app.py

app.py

@@ -638,13 +638,14 @@ def analyze_audio_segments(y, sr, model, segment_duration=10):
 # HYBRID APNEA DETECTION - NEW!
 # ========================================
 # ========================================
-# HYBRID APNEA DETECTION - RELAXED VERSION
+# HYBRID APNEA DETECTION - ONLY BETWEEN SNORING
 # ========================================
 def count_precise_apneas(y, sr, segment_results, min_apnea=10, max_apnea=90):
     """
-    HYBRID APPROACH with RELAXED validation criteria
-    Stage 1: Quick energy-based detection (1-second resolution)
-    Stage 2: Validate candidates with ML model (RELAXED thresholds)
+    HYBRID APPROACH - Detects apnea ONLY between snoring periods
+    Stage 1: Use segment_results to find confirmed snoring periods
+    Stage 2: Find gaps between snoring periods
+    Stage 3: Validate gaps with ML model
     
     Risk Levels:
     - 10-15 seconds: LOW
@@ -653,200 +654,142 @@ def count_precise_apneas(y, sr, segment_results, min_apnea=10, max_apnea=90):
     
     Returns: (apnea_count, apnea_events_for_annotations)
     """
-    print("\n🎯 HYBRID APNEA DETECTION (RELAXED)")
-    print("   Stage 1: Fast activity screening...")
-    
-    # STAGE 1: Quick activity detection at 1-second intervals
-    frame_duration = 1.0
-    frame_length = int(frame_duration * sr)
-    num_frames = int(np.ceil(len(y) / frame_length))
-    
-    timeline = []
-    for i in range(num_frames):
-        start = i * frame_length
-        end = min(start + frame_length, len(y))
-        chunk = y[start:end]
-        
-        if len(chunk) < frame_length // 2:
-            continue
-        
-        # RELAXED: More lenient energy check
-        rms = np.sqrt(np.mean(chunk**2)) if len(chunk) > 0 else 0
-        
-        # Simple zero-crossing rate
-        zcr = np.sum(np.abs(np.diff(np.sign(chunk)))) / (2 * len(chunk)) if len(chunk) > 0 else 0
-        
-        # RELAXED activity detection: lower threshold
-        is_active = (
-            rms > 0.008 and     # RELAXED from 0.01 (more sensitive to quiet snoring)
-            rms < 0.6 and       # RELAXED from 0.5 (allow louder sounds)
-            zcr < 0.18          # RELAXED from 0.15 (more tolerant)
-        )
-        
-        timeline.append({
-            'time': i * frame_duration,
-            'active': is_active,
-            'energy': rms
-        })
+    print("\n🎯 APNEA DETECTION (Only between SNORING)")
+    print("   Stage 1: Identifying confirmed snoring periods...")
+    
+    # STAGE 1: Get all confirmed snoring segments from segment_results
+    snoring_periods = []
+    
+    for seg in segment_results:
+        if seg['is_snoring']:  # Only HIGH and MEDIUM confidence snores
+            snoring_periods.append({
+                'start': seg['start_time'],
+                'end': seg['end_time'],
+                'confidence': seg['confidence_level']
+            })
     
-    active_count = sum(1 for t in timeline if t['active'])
-    print(f"      Active: {active_count}/{len(timeline)} seconds")
+    print(f"      Found {len(snoring_periods)} snoring periods")
     
-    # STAGE 2: Find gaps - RELAXED validation
-    print("   Stage 2: Finding and validating gaps...")
+    if len(snoring_periods) < 2:
+        print("      ⚠️ Need at least 2 snoring periods to detect apnea")
+        return 0, []
+    
+    # STAGE 2: Find gaps BETWEEN snoring periods
+    print("   Stage 2: Finding gaps between snoring periods...")
     
     apnea_candidates = []
-    in_gap = False
-    gap_start = None
-    last_active_time = 0
     
-    for i in range(len(timeline)):
-        current = timeline[i]
+    for i in range(len(snoring_periods) - 1):
+        current_snore = snoring_periods[i]
+        next_snore = snoring_periods[i + 1]
         
-        if current['active']:
-            # Active period
-            if in_gap and gap_start is not None:
-                # Gap just ended
-                gap_duration = current['time'] - gap_start
-                
-                if min_apnea <= gap_duration <= max_apnea:
-                    apnea_candidates.append({
-                        'start': gap_start,
-                        'end': current['time'],
-                        'duration': gap_duration,
-                        'before_time': last_active_time,
-                        'after_time': current['time']
-                    })
-                
-                in_gap = False
-                gap_start = None
-            
-            last_active_time = current['time']
-        else:
-            # Inactive period
-            if not in_gap:
-                gap_start = current['time']
-                in_gap = True
+        # Calculate gap between end of current snore and start of next snore
+        gap_start = current_snore['end']
+        gap_end = next_snore['start']
+        gap_duration = gap_end - gap_start
+        
+        # Only consider gaps in apnea range
+        if min_apnea <= gap_duration <= max_apnea:
+            apnea_candidates.append({
+                'start': gap_start,
+                'end': gap_end,
+                'duration': gap_duration,
+                'before_snore': current_snore,
+                'after_snore': next_snore
+            })
+            print(f"      Found gap: {gap_start:.1f}s → {gap_end:.1f}s (duration: {gap_duration:.1f}s)")
     
     print(f"      Found {len(apnea_candidates)} potential apnea candidates")
     
-    # STAGE 3: Validate with ML model - RELAXED THRESHOLDS
+    # STAGE 3: Validate with ML model
     validated_apneas = []
     
-    if len(apnea_candidates) > 0 and finetuned_snore_model is not None:
-        print("   Stage 3: ML validation (RELAXED thresholds)...")
-        
-        for idx, candidate in enumerate(apnea_candidates):
-            # Extract audio segments around the gap
-            buffer = 5.0  # 5 seconds before/after
-            
-            # Before segment (should be snoring)
-            before_start = max(0, candidate['start'] - buffer)
-            before_end = candidate['start']
-            before_start_sample = int(before_start * sr)
-            before_end_sample = int(before_end * sr)
-            before_audio = y[before_start_sample:before_end_sample]
-            
-            # After segment (should be snoring)
-            after_start = candidate['end']
-            after_end = min(len(y)/sr, candidate['end'] + buffer)
-            after_start_sample = int(after_start * sr)
-            after_end_sample = int(after_end * sr)
-            after_audio = y[after_start_sample:after_end_sample]
-            
-            # Gap segment (should NOT be snoring)
-            gap_start_sample = int(candidate['start'] * sr)
-            gap_end_sample = int(candidate['end'] * sr)
-            gap_audio = y[gap_start_sample:gap_end_sample]
+    if len(apnea_candidates) > 0:
+        if finetuned_snore_model is not None:
+            print("   Stage 3: ML validation (confirming silence in gap)...")
             
-            # Validate with model
-            try:
-                # Resample to 22050 for model
-                if sr != SAMPLE_RATE:
-                    before_audio = librosa.resample(before_audio, orig_sr=sr, target_sr=SAMPLE_RATE)
-                    after_audio = librosa.resample(after_audio, orig_sr=sr, target_sr=SAMPLE_RATE)
-                    gap_audio = librosa.resample(gap_audio, orig_sr=sr, target_sr=SAMPLE_RATE)
+            for idx, candidate in enumerate(apnea_candidates):
+                # We already KNOW there's snoring before and after (from segment_results)
+                # We just need to confirm the GAP is actually silent
                 
-                # Get predictions
-                before_prob = predict_snoring_segment(before_audio, SAMPLE_RATE, finetuned_snore_model) if len(before_audio) > sr else 0
-                after_prob = predict_snoring_segment(after_audio, SAMPLE_RATE, finetuned_snore_model) if len(after_audio) > sr else 0
-                gap_prob = predict_snoring_segment(gap_audio, SAMPLE_RATE, finetuned_snore_model) if len(gap_audio) > sr else 0
+                # Extract gap audio
+                gap_start_sample = int(candidate['start'] * sr)
+                gap_end_sample = int(candidate['end'] * sr)
+                gap_audio = y[gap_start_sample:gap_end_sample]
                 
-                # RELAXED validation thresholds
-                snoring_threshold = 0.35  # RELAXED from 0.50 (accept more borderline cases)
-                silence_threshold = 0.50  # RELAXED from 0.40 (accept higher gap values)
-                
-                # RELAXED validation logic - THREE scenarios
-                
-                # Scenario 1: IDEAL - Both sides snoring, gap is silent
-                ideal_pattern = (
-                    before_prob > snoring_threshold and
-                    after_prob > snoring_threshold and
-                    gap_prob < silence_threshold
-                )
-                
-                # Scenario 2: ACCEPTABLE - At least ONE side snoring, gap is silent
-                acceptable_pattern = (
-                    (before_prob > snoring_threshold or after_prob > snoring_threshold) and
-                    gap_prob < silence_threshold
-                )
-                
-                # Scenario 3: MINIMAL - Just the gap is clearly silent (energy-based detection was good enough)
-                minimal_pattern = (
-                    gap_prob < silence_threshold - 0.1  # Very low gap probability (< 0.4)
-                )
-                
-                # Accept if ANY scenario matches
-                is_valid = ideal_pattern or acceptable_pattern or minimal_pattern
-                
-                if is_valid:
-                    # Determine confidence level
-                    if ideal_pattern:
-                        confidence_type = "HIGH"
-                    elif acceptable_pattern:
-                        confidence_type = "MEDIUM"
-                    else:
-                        confidence_type = "LOW"
+                try:
+                    # Resample to 22050 for model
+                    if sr != SAMPLE_RATE:
+                        gap_audio = librosa.resample(gap_audio, orig_sr=sr, target_sr=SAMPLE_RATE)
                     
-                    # NEW RISK LEVEL ASSIGNMENT
-                    duration = candidate['duration']
-                    if duration < 15:
-                        risk_level = "LOW"
-                    elif duration < 20:
-                        risk_level = "MEDIUM"
-                    else:  # >= 20 seconds
-                        risk_level = "HIGH"
+                    # Validate that the gap is NOT snoring (should be silent)
+                    gap_prob = predict_snoring_segment(gap_audio, SAMPLE_RATE, finetuned_snore_model) if len(gap_audio) > sr else 0
                     
-                    validated_apneas.append({
-                        'label': 'APNEA',
-                        'start_sec': round(candidate['start'], 2),
-                        'end_sec': round(candidate['end'], 2),
-                        'duration': round(candidate['duration'], 2),
-                        'risk_level': risk_level,  # NEW: Updated risk levels
-                        'confidence': {
-                            'before_snoring': round(before_prob, 3),
-                            'gap_silence': round(1 - gap_prob, 3),
-                            'after_snoring': round(after_prob, 3),
-                            'validation_type': confidence_type
-                        }
-                    })
-                    print(f"      ✅ VALIDATED ({confidence_type}): {candidate['start']:.1f}s → {candidate['end']:.1f}s "
-                          f"(Before: {before_prob:.2f}, Gap: {gap_prob:.2f}, After: {after_prob:.2f}) - Risk: {risk_level}")
-                else:
-                    print(f"      ❌ REJECTED: {candidate['start']:.1f}s → {candidate['end']:.1f}s "
-                          f"(Before: {before_prob:.2f}, Gap: {gap_prob:.2f}, After: {after_prob:.2f})")
+                    # RELAXED threshold: gap should have low snoring probability
+                    silence_threshold = 0.50  # Gap should be < 0.50 snoring probability
                     
-            except Exception as e:
-                print(f"      ⚠️ Validation error for candidate {idx+1}: {e}")
-                
-                # RELAXED: On error, accept the candidate anyway (energy-based detection was good)
-                # NEW RISK LEVEL ASSIGNMENT
+                    is_silent = gap_prob < silence_threshold
+                    
+                    if is_silent:
+                        # Determine risk level
+                        duration = candidate['duration']
+                        if duration < 15:
+                            risk_level = "LOW"
+                        elif duration < 20:
+                            risk_level = "MEDIUM"
+                        else:  # >= 20 seconds
+                            risk_level = "HIGH"
+                        
+                        validated_apneas.append({
+                            'label': 'APNEA',
+                            'start_sec': round(candidate['start'], 2),
+                            'end_sec': round(candidate['end'], 2),
+                            'duration': round(candidate['duration'], 2),
+                            'risk_level': risk_level,
+                            'confidence': {
+                                'before_snoring': candidate['before_snore']['confidence'],
+                                'gap_silence': round(1 - gap_prob, 3),
+                                'after_snoring': candidate['after_snore']['confidence']
+                            }
+                        })
+                        print(f"      ✅ VALIDATED: {candidate['start']:.1f}s → {candidate['end']:.1f}s "
+                              f"(Gap silence: {gap_prob:.2f}) - Risk: {risk_level}")
+                    else:
+                        print(f"      ❌ REJECTED: {candidate['start']:.1f}s → {candidate['end']:.1f}s "
+                              f"(Gap has snoring: {gap_prob:.2f})")
+                        
+                except Exception as e:
+                    print(f"      ⚠️ Validation error for candidate {idx+1}: {e}")
+                    
+                    # On error, be conservative - still accept if gap is long enough
+                    if candidate['duration'] >= 15:
+                        duration = candidate['duration']
+                        if duration < 15:
+                            risk_level = "LOW"
+                        elif duration < 20:
+                            risk_level = "MEDIUM"
+                        else:
+                            risk_level = "HIGH"
+                        
+                        validated_apneas.append({
+                            'label': 'APNEA',
+                            'start_sec': round(candidate['start'], 2),
+                            'end_sec': round(candidate['end'], 2),
+                            'duration': round(candidate['duration'], 2),
+                            'risk_level': risk_level
+                        })
+                        print(f"      ⚠️ ACCEPTED (validation skipped): {candidate['start']:.1f}s → {candidate['end']:.1f}s - Risk: {risk_level}")
+                    continue
+        else:
+            # No model - accept all candidates between snoring periods
+            print("   Stage 3: No ML model, accepting all gaps between snoring...")
+            for candidate in apnea_candidates:
                 duration = candidate['duration']
                 if duration < 15:
                     risk_level = "LOW"
                 elif duration < 20:
                     risk_level = "MEDIUM"
-                else:  # >= 20 seconds
+                else:
                     risk_level = "HIGH"
                 
                 validated_apneas.append({
@@ -854,30 +797,8 @@ def count_precise_apneas(y, sr, segment_results, min_apnea=10, max_apnea=90):
                     'start_sec': round(candidate['start'], 2),
                     'end_sec': round(candidate['end'], 2),
                     'duration': round(candidate['duration'], 2),
-                    'risk_level': risk_level  # NEW: Updated risk levels
+                    'risk_level': risk_level
                 })
-                print(f"      ⚠️ ACCEPTED (validation skipped): {candidate['start']:.1f}s → {candidate['end']:.1f}s - Risk: {risk_level}")
-                continue
-    else:
-        # No model validation - ACCEPT all energy-based candidates
-        print("   Stage 3: Skipping ML validation, accepting all candidates...")
-        for candidate in apnea_candidates:
-            # NEW RISK LEVEL ASSIGNMENT
-            duration = candidate['duration']
-            if duration < 15:
-                risk_level = "LOW"
-            elif duration < 20:
-                risk_level = "MEDIUM"
-            else:  # >= 20 seconds
-                risk_level = "HIGH"
-            
-            validated_apneas.append({
-                'label': 'APNEA',
-                'start_sec': round(candidate['start'], 2),
-                'end_sec': round(candidate['end'], 2),
-                'duration': round(candidate['duration'], 2),
-                'risk_level': risk_level  # NEW: Updated risk levels
-            })
     
     print(f"   ✅ Final: {len(validated_apneas)} validated apnea events\n")