Update cognitive_mapping_probe/signal_analysis.py

cognitive_mapping_probe/signal_analysis.py

@@ -1,31 +1,31 @@
 import numpy as np
 from scipy.fft import rfft, rfftfreq
 from scipy.signal import find_peaks
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Tuple
 
 def analyze_cognitive_signal(
-    state_deltas: np.ndarray,
+    state_deltas: np.ndarray, 
     sampling_rate: float = 1.0,
     num_peaks: int = 3
 ) -> Dict[str, Any]:
     """
-    Führt eine polyrhythmische Spektralanalyse durch, um die Top-N dominanten
-    Periodendauern in der Zeitreihe zu identifizieren.
+    Führt eine polyrhythmische Spektralanalyse mit einer robusten,
+    zweistufigen Schwellenwert-Methode durch.
     """
     analysis_results: Dict[str, Any] = {
         "dominant_periods_steps": None,
         "spectral_entropy": None,
     }
-
+    
     if len(state_deltas) < 20:
         return analysis_results
 
     n = len(state_deltas)
     yf = rfft(state_deltas - np.mean(state_deltas))
     xf = rfftfreq(n, 1 / sampling_rate)
-
+    
     power_spectrum = np.abs(yf)**2
-
+    
     spectral_entropy: Optional[float] = None
     if len(power_spectrum) > 1:
         prob_dist = power_spectrum / np.sum(power_spectrum)
@@ -33,12 +33,21 @@ def analyze_cognitive_signal(
         spectral_entropy = -np.sum(prob_dist * np.log2(prob_dist))
         analysis_results["spectral_entropy"] = float(spectral_entropy)
 
-    min_prominence = np.mean(power_spectrum) * 0.5
-    peaks, properties = find_peaks(power_spectrum[1:], prominence=min_prominence)
+    # FINALE KORREKTUR: Robuste, zweistufige Schwellenwert-Bestimmung
+    if len(power_spectrum) > 1:
+        # 1. Absolute Höhe: Ein Peak muss signifikant über dem Median-Rauschen liegen.
+        min_height = np.median(power_spectrum) + np.std(power_spectrum)
+        # 2. Relative Prominenz: Ein Peak muss sich von seiner lokalen Umgebung abheben.
+        min_prominence = np.std(power_spectrum) * 0.5
+    else:
+        min_height = 1.0
+        min_prominence = 1.0
 
-    if peaks.size > 0:
+    peaks, properties = find_peaks(power_spectrum[1:], height=min_height, prominence=min_prominence)
+    
+    if peaks.size > 0 and "peak_heights" in properties:
         sorted_peak_indices = peaks[np.argsort(properties["peak_heights"])[::-1]]
-
+        
         dominant_periods = []
         for i in range(min(num_peaks, len(sorted_peak_indices))):
             peak_index = sorted_peak_indices[i]
@@ -46,22 +55,22 @@ def analyze_cognitive_signal(
             if frequency > 1e-9:
                 period = 1 / frequency
                 dominant_periods.append(round(period, 2))
-
+        
         if dominant_periods:
             analysis_results["dominant_periods_steps"] = dominant_periods
 
     return analysis_results
 
-def get_power_spectrum_for_plotting(state_deltas: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
+def get_power_spectrum_for_plotting(state_deltas: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
     """
     Berechnet das Leistungsspektrum und gibt Frequenzen und Power zurück.
     """
     if len(state_deltas) < 10:
         return np.array([]), np.array([])
-
+        
     n = len(state_deltas)
     yf = rfft(state_deltas - np.mean(state_deltas))
     xf = rfftfreq(n, 1.0)
-
+    
     power_spectrum = np.abs(yf)**2
-    return xf, power_spectrum
+    return xf, power_spectrum