frequency_response_analyzer.py

#!/usr/bin/env python3
"""
Frequenzganganalyse-System für Mikrofon-Qualitätsbewertung
Vergleicht das gesendete Chirp-Signal mit der Mikrofonaufnahme
"""

import numpy as np
import matplotlib.pyplot as plt
from scipy import signal
from scipy.fft import fft, fftfreq
import tkinter as tk
from tkinter import ttk, messagebox
import threading
import time
import pyaudio
import wave
import io


class FrequencyResponseAnalyzer:
    """Analysiert den Frequenzgang von Mikrofonen durch Chirp-Signal-Vergleich"""
    
    def __init__(self, sample_rate=44100, duration=2.0, f_start=20, f_end=20000):
        self.sample_rate = sample_rate
        self.duration = duration
        self.f_start = f_start
        self.f_end = f_end
        
        # Generiere Referenz-Chirp
        self.reference_chirp = self._generate_chirp()
        
        # Analyseergebnisse
        self.analysis_results = {}
        
    def _generate_chirp(self):
        """Generiert ein lineares Chirp-Signal für Frequenzgangmessung"""
        t = np.linspace(0, self.duration, int(self.sample_rate * self.duration))
        
        # Lineares Chirp von f_start bis f_end
        chirp = signal.chirp(t, self.f_start, self.duration, self.f_end, method='linear')
        
        # Fensterung für saubere Start/Ende
        window = signal.windows.tukey(len(chirp), alpha=0.1)
        chirp = chirp * window
        
        return chirp.astype(np.float32)
    
    def generate_chirp_with_silence(self, silence_before=0.5, silence_after=0.5):
        """Generiert Chirp mit Stille davor und danach für bessere Messung"""
        silence_samples_before = int(self.sample_rate * silence_before)
        silence_samples_after = int(self.sample_rate * silence_after)
        
        silence_start = np.zeros(silence_samples_before, dtype=np.float32)
        silence_end = np.zeros(silence_samples_after, dtype=np.float32)
        
        full_signal = np.concatenate([silence_start, self.reference_chirp, silence_end])
        return full_signal
    
    def analyze_recorded_signal(self, recorded_signal, mic_name="Unknown"):
        """Analysiert aufgenommenes Signal gegen Referenz-Chirp"""
        try:
            # Zeitversatz durch Kreuzkorrelation finden
            correlation = signal.correlate(recorded_signal, self.reference_chirp, mode='full')
            delay_samples = np.argmax(correlation) - len(self.reference_chirp) + 1
            
            # Signal entsprechend dem Delay ausrichten
            if delay_samples > 0:
                aligned_signal = recorded_signal[delay_samples:delay_samples + len(self.reference_chirp)]
            else:
                aligned_signal = recorded_signal[:len(self.reference_chirp)]
            
            # Auf gleiche Länge bringen
            min_len = min(len(aligned_signal), len(self.reference_chirp))
            aligned_signal = aligned_signal[:min_len]
            reference_signal = self.reference_chirp[:min_len]
            
            # FFT für beide Signale
            ref_fft = fft(reference_signal)
            rec_fft = fft(aligned_signal)
            
            # Frequenzvektor
            freqs = fftfreq(len(ref_fft), 1/self.sample_rate)
            
            # Nur positive Frequenzen
            positive_freq_mask = freqs >= 0
            freqs = freqs[positive_freq_mask]
            ref_fft = ref_fft[positive_freq_mask]
            rec_fft = rec_fft[positive_freq_mask]
            
            # Frequenzgang berechnen (Magnitude)
            ref_magnitude = np.abs(ref_fft)
            rec_magnitude = np.abs(rec_fft)
            
            # Frequenzgang-Verhältnis (in dB)
            with np.errstate(divide='ignore', invalid='ignore'):
                frequency_response = 20 * np.log10(rec_magnitude / (ref_magnitude + 1e-10))
                frequency_response[~np.isfinite(frequency_response)] = -60.0
            
            # Phasengang
            ref_phase = np.angle(ref_fft)
            rec_phase = np.angle(rec_fft)
            phase_difference = rec_phase - ref_phase
            
            # Phase unwrapping
            phase_difference = np.unwrap(phase_difference)
            
            # Qualitätsmetriken berechnen
            quality_metrics = self._calculate_quality_metrics(
                freqs, frequency_response, phase_difference, 
                reference_signal, aligned_signal
            )
            
            # Ergebnisse speichern
            result = {
                'mic_name': mic_name,
                'frequencies': freqs,
                'frequency_response': frequency_response,
                'phase_difference': phase_difference,
                'reference_signal': reference_signal,
                'recorded_signal': aligned_signal,
                'delay_samples': delay_samples,
                'delay_ms': (delay_samples / self.sample_rate) * 1000,
                'quality_metrics': quality_metrics,
                'timestamp': time.time()
            }
            
            self.analysis_results[mic_name] = result
            return result
            
        except Exception as e:
            print(f"Fehler bei der Frequenzganganalyse: {e}")
            return None
    
    def _calculate_quality_metrics(self, freqs, freq_response, phase_diff, ref_signal, rec_signal):
        """Berechnet Qualitätsmetriken für den Frequenzgang"""
        metrics = {}
        
        # Frequenzbänder definieren
        bands = {
            'bass': (20, 250),
            'low_mid': (250, 500),
            'mid': (500, 2000),
            'high_mid': (2000, 4000),
            'treble': (4000, 8000),
            'high_treble': (8000, 20000)
        }
        
        # Metriken pro Frequenzband
        for band_name, (f_low, f_high) in bands.items():
            mask = (freqs >= f_low) & (freqs <= f_high)
            if np.any(mask):
                band_response = freq_response[mask]
                metrics[f'{band_name}_mean_db'] = np.mean(band_response)
                metrics[f'{band_name}_std_db'] = np.std(band_response)
                metrics[f'{band_name}_min_db'] = np.min(band_response)
                metrics[f'{band_name}_max_db'] = np.max(band_response)
        
        # Gesamtmetriken
        speech_band = (300, 3400)  # Typischer Sprachbereich
        speech_mask = (freqs >= speech_band[0]) & (freqs <= speech_band[1])
        
        if np.any(speech_mask):
            speech_response = freq_response[speech_mask]
            metrics['speech_band_mean_db'] = np.mean(speech_response)
            metrics['speech_band_flatness'] = np.std(speech_response)  # Niedriger = flacher
            metrics['speech_band_range_db'] = np.max(speech_response) - np.min(speech_response)
        
        # SNR und THD+N Schätzung
        rms_ref = np.sqrt(np.mean(ref_signal**2))
        rms_rec = np.sqrt(np.mean(rec_signal**2))
        
        if rms_ref > 0:
            metrics['amplitude_ratio_db'] = 20 * np.log10(rms_rec / rms_ref)
        
        # Korrelationskoeffizient zwischen Referenz und Aufnahme
        correlation_coeff = np.corrcoef(ref_signal, rec_signal)[0, 1]
        metrics['correlation_coefficient'] = correlation_coeff
        metrics['correlation_quality'] = 'Excellent' if correlation_coeff > 0.9 else \
                                        'Good' if correlation_coeff > 0.7 else \
                                        'Fair' if correlation_coeff > 0.5 else 'Poor'
        
        # Frequenzgang-Glätte (Standardabweichung)
        metrics['overall_flatness'] = np.std(freq_response)
        metrics['flatness_quality'] = 'Excellent' if metrics['overall_flatness'] < 3.0 else \
                                     'Good' if metrics['overall_flatness'] < 6.0 else \
                                     'Fair' if metrics['overall_flatness'] < 10.0 else 'Poor'
        
        return metrics
    
    def plot_frequency_response(self, mic_names=None, save_path=None):
        """Plottet Frequenzgänge für ausgewählte Mikrofone"""
        if not self.analysis_results:
            print("Keine Analyseergebnisse vorhanden")
            return
        
        if mic_names is None:
            mic_names = list(self.analysis_results.keys())
        
        fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
        
        colors = plt.cm.tab10(np.linspace(0, 1, len(mic_names)))
        
        for i, mic_name in enumerate(mic_names):
            if mic_name not in self.analysis_results:
                continue
                
            result = self.analysis_results[mic_name]
            freqs = result['frequencies']
            freq_response = result['frequency_response']
            phase_diff = result['phase_difference']
            
            # Frequenzgang (Magnitude)
            ax1.semilogx(freqs, freq_response, color=colors[i], label=mic_name, linewidth=2)
            
            # Phasengang
            ax2.semilogx(freqs, np.degrees(phase_diff), color=colors[i], label=mic_name, linewidth=2)
        
        # Frequenzgang-Plot formatieren
        ax1.set_xlabel('Frequenz [Hz]')
        ax1.set_ylabel('Magnitude [dB]')
        ax1.set_title('Frequenzgang-Analyse: Mikrofon vs. Referenz')
        ax1.grid(True, alpha=0.3)
        ax1.legend()
        ax1.set_xlim(20, 20000)
        ax1.axhline(y=0, color='r', linestyle='--', alpha=0.5, label='Referenz')
        ax1.axhline(y=-3, color='orange', linestyle='--', alpha=0.5, label='-3dB')
        ax1.axhline(y=3, color='orange', linestyle='--', alpha=0.5, label='+3dB')
        
        # Phasengang-Plot formatieren
        ax2.set_xlabel('Frequenz [Hz]')
        ax2.set_ylabel('Phasendifferenz [°]')
        ax2.set_title('Phasengang-Analyse')
        ax2.grid(True, alpha=0.3)
        ax2.legend()
        ax2.set_xlim(20, 20000)
        
        plt.tight_layout()
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
        
        plt.show()
    
    def generate_quality_report(self, mic_name):
        """Generiert einen detaillierten Qualitätsbericht"""
        if mic_name not in self.analysis_results:
            return "Keine Analyseergebnisse für dieses Mikrofon vorhanden."
        
        result = self.analysis_results[mic_name]
        metrics = result['quality_metrics']
        
        report = f"""
🎤 FREQUENZGANG-ANALYSE: {mic_name}
{'='*50}

📊 GESAMTBEWERTUNG:
   Korrelation mit Referenz: {metrics['correlation_coefficient']:.3f} ({metrics['correlation_quality']})
   Frequenzgang-Glätte: {metrics['overall_flatness']:.1f}dB ({metrics['flatness_quality']})
   Zeitverzögerung: {result['delay_ms']:.1f}ms

🔊 SPRACHBEREICH (300-3400 Hz):
   Durchschnittliche Abweichung: {metrics.get('speech_band_mean_db', 0):.1f}dB
   Frequenzgang-Variation: {metrics.get('speech_band_flatness', 0):.1f}dB
   Dynamikbereich: {metrics.get('speech_band_range_db', 0):.1f}dB

🎵 FREQUENZBÄNDER:
   Bass (20-250Hz): {metrics.get('bass_mean_db', 0):.1f}dB (±{metrics.get('bass_std_db', 0):.1f}dB)
   Tiefen Mitten (250-500Hz): {metrics.get('low_mid_mean_db', 0):.1f}dB (±{metrics.get('low_mid_std_db', 0):.1f}dB)
   Mitten (500-2000Hz): {metrics.get('mid_mean_db', 0):.1f}dB (±{metrics.get('mid_std_db', 0):.1f}dB)
   Hohe Mitten (2-4kHz): {metrics.get('high_mid_mean_db', 0):.1f}dB (±{metrics.get('high_mid_std_db', 0):.1f}dB)
   Höhen (4-8kHz): {metrics.get('treble_mean_db', 0):.1f}dB (±{metrics.get('treble_std_db', 0):.1f}dB)
   Superhöhen (8-20kHz): {metrics.get('high_treble_mean_db', 0):.1f}dB (±{metrics.get('high_treble_std_db', 0):.1f}dB)

📈 EMPFEHLUNGEN:
"""
        
        # Empfehlungen basierend auf Metriken
        if metrics.get('speech_band_flatness', 0) > 5.0:
            report += "   ⚠️ Ungleichmäßiger Frequenzgang im Sprachbereich - EQ empfohlen\n"
        
        if metrics.get('correlation_coefficient', 0) < 0.7:
            report += "   ⚠️ Niedrige Korrelation - Mikrofonposition oder -qualität prüfen\n"
        
        if abs(metrics.get('speech_band_mean_db', 0)) > 6.0:
            report += "   ⚠️ Signifikante Pegeländerung im Sprachbereich - Verstärkung anpassen\n"
        
        if result['delay_ms'] > 50:
            report += "   ⚠️ Hohe Latenz - Audio-Interface oder Puffergrößen überprüfen\n"
        
        if metrics.get('correlation_coefficient', 0) > 0.9 and metrics.get('overall_flatness', 0) < 3.0:
            report += "   ✅ Ausgezeichnete Mikrofonqualität!\n"
        
        return report


class FrequencyResponseGUI:
    """GUI für die Frequenzganganalyse"""
    
    def __init__(self, root):
        self.root = root
        self.root.title("🔬 Frequenzgang-Analyse für Mikrofone")
        self.root.geometry("900x700")
        
        self.analyzer = FrequencyResponseAnalyzer()
        self.audio = pyaudio.PyAudio()
        
        self.setup_ui()
        
    def setup_ui(self):
        """UI aufbauen"""
        # Titel
        title = tk.Label(self.root, text="🔬 Frequenzgang-Analyse für Mikrofone", 
                        font=('Arial', 16, 'bold'), fg='blue')
        title.pack(pady=10)
        
        # Kontrollen
        control_frame = tk.LabelFrame(self.root, text="🎛️ Steuerung", font=('Arial', 12, 'bold'))
        control_frame.pack(pady=5, padx=10, fill=tk.X)
        
        # Parameter-Frame
        param_frame = tk.Frame(control_frame)
        param_frame.pack(fill=tk.X, padx=5, pady=5)
        
        tk.Label(param_frame, text="Dauer:").grid(row=0, column=0, sticky='w')
        self.duration_var = tk.DoubleVar(value=2.0)
        tk.Spinbox(param_frame, from_=1.0, to=5.0, increment=0.5, 
                  textvariable=self.duration_var, width=10).grid(row=0, column=1, padx=5)
        
        tk.Label(param_frame, text="Start-Freq:").grid(row=0, column=2, sticky='w')
        self.f_start_var = tk.IntVar(value=20)
        tk.Spinbox(param_frame, from_=20, to=1000, increment=10,
                  textvariable=self.f_start_var, width=10).grid(row=0, column=3, padx=5)
        
        tk.Label(param_frame, text="End-Freq:").grid(row=0, column=4, sticky='w')
        self.f_end_var = tk.IntVar(value=20000)
        tk.Spinbox(param_frame, from_=1000, to=20000, increment=1000,
                  textvariable=self.f_end_var, width=10).grid(row=0, column=5, padx=5)
        
        # Buttons
        button_frame = tk.Frame(control_frame)
        button_frame.pack(fill=tk.X, padx=5, pady=5)
        
        tk.Button(button_frame, text="🔊 Chirp abspielen", 
                 command=self.play_chirp, bg='lightgreen').pack(side=tk.LEFT, padx=5)
        
        tk.Button(button_frame, text="🎤 Aufnahme & Analyse", 
                 command=self.record_and_analyze, bg='lightblue').pack(side=tk.LEFT, padx=5)
        
        tk.Button(button_frame, text="📊 Plot zeigen", 
                 command=self.show_plot, bg='lightyellow').pack(side=tk.LEFT, padx=5)
        
        tk.Button(button_frame, text="📋 Bericht", 
                 command=self.show_report, bg='lightcoral').pack(side=tk.LEFT, padx=5)
        
        # Mikrofon-Liste
        mic_frame = tk.LabelFrame(self.root, text="🎤 Analysierte Mikrofone")
        mic_frame.pack(pady=5, padx=10, fill=tk.BOTH, expand=False)
        
        self.mic_listbox = tk.Listbox(mic_frame, height=6)
        self.mic_listbox.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
        
        # Status und Ergebnisse
        result_frame = tk.LabelFrame(self.root, text="📊 Analyseergebnisse")
        result_frame.pack(pady=5, padx=10, fill=tk.BOTH, expand=True)
        
        self.result_text = tk.Text(result_frame, font=('Courier', 10))
        scrollbar = tk.Scrollbar(result_frame, orient="vertical", command=self.result_text.yview)
        self.result_text.configure(yscrollcommand=scrollbar.set)
        
        self.result_text.pack(side="left", fill="both", expand=True, padx=5, pady=5)
        scrollbar.pack(side="right", fill="y")
        
        # Initiale Meldung
        self.log_message("🎤 Frequenzgang-Analysesystem bereit")
        self.log_message("💡 Schritt 1: Chirp-Signal abspielen")
        self.log_message("💡 Schritt 2: Aufnahme & Analyse starten")
    
    def log_message(self, message):
        """Nachricht in Textfeld einfügen"""
        self.result_text.insert(tk.END, f"[{time.strftime('%H:%M:%S')}] {message}\n")
        self.result_text.see(tk.END)
        self.root.update_idletasks()
    
    def play_chirp(self):
        """Spielt das Chirp-Signal ab"""
        try:
            self.log_message("🔊 Spiele Chirp-Signal ab...")
            
            # Parameter aktualisieren
            self.analyzer.duration = self.duration_var.get()
            self.analyzer.f_start = self.f_start_var.get()
            self.analyzer.f_end = self.f_end_var.get()
            self.analyzer.reference_chirp = self.analyzer._generate_chirp()
            
            # Chirp mit Stille
            playback_signal = self.analyzer.generate_chirp_with_silence()
            
            # PyAudio Stream für Wiedergabe
            stream = self.audio.open(
                format=pyaudio.paFloat32,
                channels=1,
                rate=self.analyzer.sample_rate,
                output=True,
                frames_per_buffer=1024
            )
            
            # Signal in Chunks abspielen
            chunk_size = 1024
            for i in range(0, len(playback_signal), chunk_size):
                chunk = playback_signal[i:i+chunk_size]
                if len(chunk) < chunk_size:
                    chunk = np.pad(chunk, (0, chunk_size - len(chunk)), 'constant')
                stream.write(chunk.tobytes())
            
            stream.stop_stream()
            stream.close()
            
            self.log_message("✅ Chirp-Signal abgespielt")
            
        except Exception as e:
            self.log_message(f"❌ Fehler beim Abspielen: {e}")
            messagebox.showerror("Fehler", f"Chirp-Wiedergabe fehlgeschlagen: {e}")
    
    def record_and_analyze(self):
        """Nimmt Audio auf und analysiert es"""
        try:
            # Mikrofon-Name abfragen
            mic_name = tk.simpledialog.askstring(
                "Mikrofonname", 
                "Name des zu testenden Mikrofons:", 
                initialvalue=f"Mikrofon_{len(self.analyzer.analysis_results) + 1}"
            )
            if not mic_name:
                return
            
            self.log_message(f"🎤 Starte Aufnahme für '{mic_name}'...")
            
            # Aufnahmedauer (etwas länger als Chirp)
            record_duration = self.duration_var.get() + 2.0  # +2s Puffer
            
            # PyAudio Stream für Aufnahme
            stream = self.audio.open(
                format=pyaudio.paFloat32,
                channels=1,
                rate=self.analyzer.sample_rate,
                input=True,
                frames_per_buffer=1024
            )
            
            self.log_message("📹 Aufnahme läuft... (sprechen Sie das Chirp-Signal)")
            
            # Aufnahme in Thread
            def record():
                frames = []
                for _ in range(0, int(self.analyzer.sample_rate / 1024 * record_duration)):
                    data = stream.read(1024)
                    frames.append(data)
                
                stream.stop_stream()
                stream.close()
                
                # Audio-Daten konvertieren
                audio_data = np.frombuffer(b''.join(frames), dtype=np.float32)
                
                # Analyse durchführen
                self.log_message("🔬 Analysiere Frequenzgang...")
                result = self.analyzer.analyze_recorded_signal(audio_data, mic_name)
                
                if result:
                    # UI aktualisieren
                    self.root.after(0, lambda: self.update_results(mic_name, result))
                else:
                    self.root.after(0, lambda: self.log_message("❌ Analyse fehlgeschlagen"))
            
            # Aufnahme-Thread starten
            record_thread = threading.Thread(target=record, daemon=True)
            record_thread.start()
            
        except Exception as e:
            self.log_message(f"❌ Aufnahmefehler: {e}")
            messagebox.showerror("Fehler", f"Aufnahme fehlgeschlagen: {e}")
    
    def update_results(self, mic_name, result):
        """Aktualisiert die Ergebnisanzeige"""
        # Mikrofon zur Liste hinzufügen
        self.mic_listbox.insert(tk.END, mic_name)
        
        # Kurze Zusammenfassung anzeigen
        metrics = result['quality_metrics']
        summary = f"""
✅ Analyse abgeschlossen: {mic_name}
   Korrelation: {metrics['correlation_coefficient']:.3f} ({metrics['correlation_quality']})
   Frequenzgang: {metrics['overall_flatness']:.1f}dB ({metrics['flatness_quality']})
   Verzögerung: {result['delay_ms']:.1f}ms
   Sprachbereich: {metrics.get('speech_band_mean_db', 0):.1f}dB Abweichung
"""
        self.log_message(summary)
    
    def show_plot(self):
        """Zeigt Frequenzgang-Plots"""
        selected_mics = [self.mic_listbox.get(i) for i in self.mic_listbox.curselection()]
        if not selected_mics:
            selected_mics = None  # Alle anzeigen
        
        self.analyzer.plot_frequency_response(selected_mics)
    
    def show_report(self):
        """Zeigt detaillierten Bericht"""
        selection = self.mic_listbox.curselection()
        if not selection:
            messagebox.showwarning("Auswahl", "Bitte ein Mikrofon auswählen")
            return
        
        mic_name = self.mic_listbox.get(selection[0])
        report = self.analyzer.generate_quality_report(mic_name)
        
        # Report-Fenster
        report_window = tk.Toplevel(self.root)
        report_window.title(f"📋 Qualitätsbericht: {mic_name}")
        report_window.geometry("800x600")
        
        report_text = tk.Text(report_window, font=('Courier', 10), wrap=tk.WORD)
        scrollbar2 = tk.Scrollbar(report_window, orient="vertical", command=report_text.yview)
        report_text.configure(yscrollcommand=scrollbar2.set)
        
        report_text.pack(side="left", fill="both", expand=True, padx=10, pady=10)
        scrollbar2.pack(side="right", fill="y")
        
        report_text.insert(tk.END, report)
    
    def __del__(self):
        """Cleanup"""
        if hasattr(self, 'audio'):
            self.audio.terminate()


if __name__ == "__main__":
    root = tk.Tk()
    app = FrequencyResponseGUI(root)
    root.mainloop()