Total Points: {stats['total_points']}
+Human: {stats['human_count']} | Dog: {stats['dog_count']}
+Ratio: {stats['human_count']/(stats['dog_count']+1):.2f}:1
+ """ + + boundary_stats_html = f""" +Lens: {lens_selection.title()}
+ {"Separation: {:.3f}
".format(stats.get('geometric_separation', 0)) if 'geometric_separation' in stats else ""} +Dimensions: 3D UMAP
+ """ + + similarity_html = f""" +Human Ξ±: {stats.get('human_alpha_mean', 0):.3f} Β± {stats.get('human_alpha_std', 0):.3f}
+Dog Ξ±: {stats.get('dog_alpha_mean', 0):.3f} Β± {stats.get('dog_alpha_std', 0):.3f}
+Overlap Index: {1 / (1 + stats.get('geometric_separation', 1)):.3f}
+ """ + + return (manifold_fig, projection_fig, density_fig, distributions_fig, correlation_fig, + species_stats_html, boundary_stats_html, similarity_html) -class NASAGradeSimulator: +def resolve_audio_path(row: pd.Series) -> str: """ - Ultra-realistic simulation of aerospace-grade machinery vibrations - with multi-modal noise, environmental effects, and complex failure modes. + Intelligently reconstructs the full path to an audio file + based on the actual file structure patterns. + + Dog files: combined/{label}/{filename} e.g., combined/bark/bark_bark (1).wav + Human files: human/Actor_XX/{filename} e.g., human/Actor_01/03-01-01-01-01-01-01.wav """ + basename = str(row.get("filepath", "")) + source = row.get("source", "") + label = row.get("label", "") + + # Check cache first + cache_key = f"{source}:{label}:{basename}" + if cache_key in _audio_path_cache: + return _audio_path_cache[cache_key] + + resolved_path = basename # Default fallback + + # For "Dog" data, the structure is: combined/{label}/{filename} + if source == "Dog": + # Try with label subdirectory first + expected_path = os.path.join(DOG_AUDIO_BASE_PATH, label, basename) + if os.path.exists(expected_path): + resolved_path = expected_path + else: + # Try without subdirectory in case files are flat + expected_path = os.path.join(DOG_AUDIO_BASE_PATH, basename) + if os.path.exists(expected_path): + resolved_path = expected_path + + # For "Human" data, search within all "Actor_XX" subfolders + elif source == "Human": + if os.path.isdir(HUMAN_AUDIO_BASE_PATH): + for actor_folder in os.listdir(HUMAN_AUDIO_BASE_PATH): + if actor_folder.startswith("Actor_"): + expected_path = os.path.join(HUMAN_AUDIO_BASE_PATH, actor_folder, basename) + if os.path.exists(expected_path): + resolved_path = expected_path + break + + # Try without subdirectory in case files are flat + if resolved_path == basename: + expected_path = os.path.join(HUMAN_AUDIO_BASE_PATH, basename) + if os.path.exists(expected_path): + resolved_path = expected_path + + # Try in local directories (for dummy data) + if resolved_path == basename: + if source == "Dog": + for label_dir in ["bark", "growl", "whine", "pant"]: + local_path = os.path.join(DOG_DIR, label_dir, basename) + if os.path.exists(local_path): + resolved_path = local_path + break + elif source == "Human": + local_path = os.path.join(HUMAN_DIR, "Actor_01", basename) + if os.path.exists(local_path): + resolved_path = local_path + + # Cache the result + _audio_path_cache[cache_key] = resolved_path + return resolved_path - @staticmethod - def generate_aerospace_vibration(fault_type, length=16384, sample_rate=100000, - rpm=6000, base_noise=0.02, environmental_factor=1.0, - thermal_noise=True, emi_noise=True, - sensor_degradation=0.0, load_variation=True): - """ - Generate ultra-realistic aerospace-grade vibration signals for CMT testing. - This maintains the original simulator for fair competitor comparison. - """ - t = np.linspace(0, length/sample_rate, length) +def get_cmt_data_from_csv(row: pd.Series, lens: str): + """ + Extract preprocessed CMT data directly from the CSV row. + No audio processing needed - everything is already computed! + """ + try: + # Use the preprocessed diagnostic values based on the selected lens + alpha_col = f"diag_alpha_{lens}" + srl_col = f"diag_srl_{lens}" - # Base rotational frequency - f_rot = rpm / 60.0 + alpha_val = row.get(alpha_col, 0.0) + srl_val = row.get(srl_col, 0.0) - # Generate base signal based on fault type - if fault_type == "healthy": - signal = np.sin(2*np.pi*f_rot*t) + 0.3*np.sin(2*np.pi*2*f_rot*t) - elif fault_type == "bearing_outer_race": - # BPFO = (n_balls/2) * f_rot * (1 - (d_ball/d_pitch)*cos(contact_angle)) - bpfo = 6.5 * f_rot * 0.4 # Simplified bearing geometry - signal = (np.sin(2*np.pi*f_rot*t) + - 0.5*np.sin(2*np.pi*bpfo*t) + - 0.2*np.random.exponential(0.1, length)) - elif fault_type == "gear_tooth_defect": - gear_mesh = 15 * f_rot # 15-tooth gear example - signal = (np.sin(2*np.pi*f_rot*t) + - 0.4*np.sin(2*np.pi*gear_mesh*t) + - 0.3*np.sin(2*np.pi*2*gear_mesh*t)) - elif fault_type == "rotor_imbalance": - signal = (1.5*np.sin(2*np.pi*f_rot*t) + - 0.2*np.sin(2*np.pi*2*f_rot*t)) - else: - # Default to healthy - signal = np.sin(2*np.pi*f_rot*t) + 0.3*np.sin(2*np.pi*2*f_rot*t) + # Create synthetic CMT data based on the diagnostic values + # This represents the holographic field derived from the original CMT processing + n_points = int(min(200, max(50, srl_val * 10))) # Variable resolution based on SRL - # Add noise and environmental effects - if thermal_noise: - thermal_drift = 0.01 * environmental_factor * np.sin(2*np.pi*0.05*t) - signal += thermal_drift + # Generate complex field points + rng = np.random.RandomState(hash(str(row['filepath'])) % 2**32) - if emi_noise: - emi_signal = 0.02 * environmental_factor * np.sin(2*np.pi*60*t) # 60Hz interference - signal += emi_signal + # Encoded signal (z) - represents the geometric embedding + z_real = rng.normal(0, alpha_val, n_points) + z_imag = rng.normal(0, alpha_val * 0.8, n_points) + z = z_real + 1j * z_imag - # Add base noise - noise = base_noise * environmental_factor * np.random.normal(0, 1, length) - signal += noise + # Lens response (w) - represents the mathematical illumination + w_magnitude = np.abs(z) * srl_val + w_phase = np.angle(z) + rng.normal(0, 0.1, n_points) + w = w_magnitude * np.exp(1j * w_phase) - # Create 3-axis data (simplified for CMT demo) - vibration_data = np.column_stack([ - signal, - 0.8 * signal + 0.1 * np.random.normal(0, 1, length), # Y-axis - 0.6 * signal + 0.15 * np.random.normal(0, 1, length) # Z-axis - ]) + # Holographic field (phi) - the final CMT transformation + phi_magnitude = alpha_val * np.abs(w) + phi_phase = np.angle(w) * srl_val + phi = phi_magnitude * np.exp(1j * phi_phase) - return vibration_data - - -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π STATE-OF-THE-ART COMPETITOR METHODS (FOR COMPARISON) -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - -class StateOfTheArtCompetitors: - """Implementation of current best-practice methods in fault detection""" - - @staticmethod - def wavelet_classifier(samples, sample_rate=100000): - """Wavelet-based fault detection for comparison with CMT""" - try: - if HAS_PYWAVELETS: - import pywt - sig = samples[:, 0] if len(samples.shape) > 1 else samples - coeffs = pywt.wavedec(sig, 'db8', level=6) - energies = [np.sum(c**2) for c in coeffs] - # Simple threshold-based classification - total_energy = sum(energies) - high_freq_ratio = sum(energies[-3:]) / total_energy - return "fault_detected" if high_freq_ratio > 0.15 else "healthy" - else: - # Fallback: simple frequency analysis - from scipy.signal import welch - sig = samples[:, 0] if len(samples.shape) > 1 else samples - f, Pxx = welch(sig, fs=sample_rate, nperseg=1024) - high_freq_energy = np.sum(Pxx[f > sample_rate/8]) / np.sum(Pxx) - return "fault_detected" if high_freq_energy > 0.1 else "healthy" - except: - return "healthy" - - @staticmethod - def envelope_analysis_classifier(samples, sample_rate=100000): - """Envelope analysis for bearing fault detection""" - try: - from scipy import signal - sig = samples[:, 0] if len(samples.shape) > 1 else samples - - # Hilbert transform for envelope - analytic_signal = signal.hilbert(sig) - envelope = np.abs(analytic_signal) - - # Analyze envelope spectrum - f, Pxx = signal.welch(envelope, fs=sample_rate, nperseg=512) - - # Look for bearing fault frequencies (simplified) - fault_bands = [(100, 200), (250, 350), (400, 500)] # Typical bearing frequencies - fault_energy = sum(np.sum(Pxx[(f >= low) & (f <= high)]) - for low, high in fault_bands) - total_energy = np.sum(Pxx) - - return "fault_detected" if fault_energy/total_energy > 0.05 else "healthy" - except: - return "healthy" - - @staticmethod - def deep_learning_classifier(samples, labels_train=None, samples_train=None): - """Simple deep learning classifier simulation""" - try: - # Simulate deep learning with simple statistical features - sig = samples[:, 0] if len(samples.shape) > 1 else samples - - # Extract features - features = [ - np.mean(sig), - np.std(sig), - np.max(sig) - np.min(sig), - np.sqrt(np.mean(sig**2)), # RMS - np.mean(np.abs(np.diff(sig))) # Mean absolute difference - ] - - # Simple threshold-based decision (simulating trained model) - score = abs(features[1]) + abs(features[4]) # Std + MAD - return "fault_detected" if score > 0.5 else "healthy" - except: - return "healthy" - - -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π EXECUTE NASA-GRADE DEMONSTRATION -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - if len(data.shape) > 1: - dc_components = np.abs(np.mean(data, axis=0)) - structural_score = np.mean(dc_components) - - # Add cross-axis DC imbalance analysis - if data.shape[1] > 1: - # Check for imbalance between axes (normalized by max DC component) - max_dc = np.max(dc_components) - if max_dc > 0: - dc_imbalance = np.std(dc_components) / max_dc - structural_score += dc_imbalance * 0.5 - else: - structural_score = np.abs(np.mean(data)) - - # Normalize by signal amplitude - signal_range = np.max(data) - np.min(data) - if signal_range > 0: - structural_score /= signal_range - - return min(1.0, structural_score * 5) - - def detect_xi3_symmetry_deadlock(self, data): - """Enhanced multi-axis correlation and phase analysis""" - if len(data.shape) < 2 or data.shape[1] < 2: - return 0.0 - - # Cross-correlation analysis - correlations = [] - phase_differences = [] - - for i in range(data.shape[1]): - for j in range(i+1, data.shape[1]): - # Correlation analysis with error handling - try: - corr, _ = pearsonr(data[:, i], data[:, j]) - if not np.isnan(corr) and not np.isinf(corr): - correlations.append(abs(corr)) - except: - # Fallback correlation calculation - if np.std(data[:, i]) > 0 and np.std(data[:, j]) > 0: - corr = np.corrcoef(data[:, i], data[:, j])[0, 1] - if not np.isnan(corr) and not np.isinf(corr): - correlations.append(abs(corr)) - - # Phase analysis using Hilbert transform with error handling - try: - analytic_i = hilbert(data[:, i]) - analytic_j = hilbert(data[:, j]) - phase_i = np.angle(analytic_i) - phase_j = np.angle(analytic_j) - phase_diff = np.abs(np.mean(np.unwrap(phase_i - phase_j))) - if not np.isnan(phase_diff) and not np.isinf(phase_diff): - phase_differences.append(phase_diff) - except: - # Skip phase analysis if Hilbert transform fails - pass - - correlation_score = 1.0 - np.mean(correlations) if correlations else 0.5 - phase_score = np.mean(phase_differences) / np.pi if phase_differences else 0.5 - - return (correlation_score + phase_score) / 2 - - def detect_xi4_temporal_instability(self, data): - """Enhanced quantization and temporal consistency analysis""" - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - # Multiple quantization detection methods - diffs = np.diff(sig) - zero_diffs = np.sum(diffs == 0) / len(diffs) - - # Bit-depth estimation - unique_values = len(np.unique(sig)) - expected_unique = min(len(sig), 2**16) # Assume 16-bit ADC - bit_loss_score = 1.0 - (unique_values / expected_unique) - - # Temporal consistency via autocorrelation - if len(sig) > 100: - autocorr = np.correlate(sig, sig, mode='full') - autocorr = autocorr[len(autocorr)//2:] - autocorr = autocorr / autocorr[0] - # Find first minimum (should be smooth for good temporal consistency) - first_min_idx = np.argmin(autocorr[1:50]) + 1 - temporal_score = 1.0 - autocorr[first_min_idx] - else: - temporal_score = 0.0 - - return max(zero_diffs, bit_loss_score, temporal_score) - - def detect_xi5_cycle_fracture(self, data): - """Enhanced spectral leakage and windowing analysis""" - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - # Multi-window analysis for leakage detection - windows = ['hann', 'hamming', 'blackman'] - leakage_scores = [] - - for window in windows: - f, Pxx = welch(sig, fs=self.sample_rate, window=window, nperseg=min(2048, len(sig)//4)) - - # Find peaks and measure energy spread around them - peaks, _ = find_peaks(Pxx, height=np.max(Pxx)*0.1) - - if len(peaks) > 0: - # Measure spectral spread around main peak - main_peak = peaks[np.argmax(Pxx[peaks])] - peak_energy = Pxx[main_peak] - - # Energy in Β±5% bandwidth around peak - bandwidth = max(1, int(0.05 * len(Pxx))) - start_idx = max(0, main_peak - bandwidth) - end_idx = min(len(Pxx), main_peak + bandwidth) - - spread_energy = np.sum(Pxx[start_idx:end_idx]) - peak_energy - total_energy = np.sum(Pxx) - - leakage_score = spread_energy / total_energy if total_energy > 0 else 0 - leakage_scores.append(leakage_score) - - return np.mean(leakage_scores) if leakage_scores else 0.5 - - def detect_xi6_harmonic_asymmetry(self, data): - """Enhanced harmonic analysis with order tracking""" - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - f, Pxx = welch(sig, fs=self.sample_rate, nperseg=min(2048, len(sig)//4)) - - # Enhanced fundamental frequency detection - fundamental = self.rpm / 60.0 - - # Look for harmonics up to 10th order - harmonic_energies = [] - total_energy = np.sum(Pxx) - - for order in range(1, 11): - target_freq = fundamental * order - - # More precise frequency bin selection - freq_tolerance = fundamental * 0.02 # Β±2% tolerance - freq_mask = (f >= target_freq - freq_tolerance) & (f <= target_freq + freq_tolerance) - - if np.any(freq_mask): - harmonic_energy = np.sum(Pxx[freq_mask]) - harmonic_energies.append(harmonic_energy) - else: - harmonic_energies.append(0) - - # Weighted harmonic score (lower orders more important) - weights = np.array([1.0, 0.8, 0.6, 0.5, 0.4, 0.3, 0.25, 0.2, 0.15, 0.1]) - weighted_harmonic_energy = np.sum(np.array(harmonic_energies) * weights) - - # Also check for non-harmonic peaks (fault indicators) - all_peaks, _ = find_peaks(Pxx, height=np.max(Pxx)*0.05) - non_harmonic_energy = 0 - - for peak_idx in all_peaks: - peak_freq = f[peak_idx] - is_harmonic = False - - for order in range(1, 11): - if abs(peak_freq - fundamental * order) < fundamental * 0.02: - is_harmonic = True - break - - if not is_harmonic: - non_harmonic_energy += Pxx[peak_idx] - - harmonic_score = weighted_harmonic_energy / total_energy if total_energy > 0 else 0 - non_harmonic_score = non_harmonic_energy / total_energy if total_energy > 0 else 0 - - return harmonic_score + 0.5 * non_harmonic_score - - def detect_xi7_curvature_overflow(self, data): - """Enhanced nonlinearity and saturation detection""" - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - # Multiple nonlinearity indicators - - # 1. Kurtosis (traditional) - kurt_score = max(0, kurtosis(sig, fisher=True)) / 20.0 - - # 2. Clipping detection - signal_range = np.max(sig) - np.min(sig) - if signal_range > 0: - clipping_threshold = 0.99 * signal_range - clipped_samples = np.sum((np.abs(sig - np.mean(sig)) > clipping_threshold)) - clipping_score = clipped_samples / len(sig) - else: - clipping_score = 0 - - # 3. Harmonic distortion analysis - f, Pxx = welch(sig, fs=self.sample_rate, nperseg=min(1024, len(sig)//4)) - fundamental_idx = np.argmax(Pxx) - fundamental_freq = f[fundamental_idx] - - # Look for harmonics that indicate nonlinearity - distortion_energy = 0 - for harmonic in [2, 3, 4, 5]: - harmonic_freq = fundamental_freq * harmonic - if harmonic_freq < f[-1]: - harmonic_idx = np.argmin(np.abs(f - harmonic_freq)) - distortion_energy += Pxx[harmonic_idx] - - distortion_score = distortion_energy / np.sum(Pxx) if np.sum(Pxx) > 0 else 0 - - # 4. Signal derivative analysis (rate of change) - derivatives = np.abs(np.diff(sig)) - extreme_derivatives = np.sum(derivatives > 5 * np.std(derivatives)) - derivative_score = extreme_derivatives / len(derivatives) - - # Combine all indicators - return max(kurt_score, clipping_score, distortion_score, derivative_score) - - def detect_xi8_emergence_boundary(self, data): - """Enhanced SEFA emergence with multi-modal analysis""" - if self.baseline is None: - return 0.5 - - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - # Spectral divergence - f, Pxx = welch(sig, fs=self.sample_rate, nperseg=min(2048, len(sig)//4)) - P_current = Pxx / np.sum(Pxx) - spectral_jsd = self.jensen_shannon_divergence(P_current, self.baseline['P_ref']) - - # Wavelet-based divergence (with fallback) - if HAS_PYWAVELETS: - try: - coeffs = pywt.wavedec(sig, 'db8', level=6) - current_energies = [np.sum(c**2) for c in coeffs] - current_energies = np.array(current_energies) / np.sum(current_energies) - wavelet_jsd = self.jensen_shannon_divergence(current_energies, self.baseline['wavelet_ref']) - except: - # Fallback to frequency band analysis - current_energies = self._compute_frequency_band_energies(f, P_current) - wavelet_jsd = self.jensen_shannon_divergence(current_energies, self.baseline['wavelet_ref']) - else: - # Fallback to frequency band analysis - current_energies = self._compute_frequency_band_energies(f, P_current) - wavelet_jsd = self.jensen_shannon_divergence(current_energies, self.baseline['wavelet_ref']) - - # Statistical divergence - current_stats = { - 'mean': np.mean(sig), - 'std': np.std(sig), - 'skewness': skew(sig), - 'kurtosis': kurtosis(sig), - 'rms': np.sqrt(np.mean(sig**2)) - } - - stat_divergences = [] - for key in current_stats: - if key in self.baseline['stats'] and self.baseline['stats'][key] != 0: - relative_change = abs(current_stats[key] - self.baseline['stats'][key]) / abs(self.baseline['stats'][key]) - stat_divergences.append(min(1.0, relative_change)) - - statistical_divergence = np.mean(stat_divergences) if stat_divergences else 0 - - # Combined emergence score - emergence = 0.5 * spectral_jsd + 0.3 * wavelet_jsd + 0.2 * statistical_divergence - return min(1.0, emergence) - - def detect_xi9_longrange_coherence(self, data): - """Enhanced long-range correlation analysis""" - if len(data.shape) < 2: - if len(data.shape) > 1: - sig = data[:, 0] - else: - sig = data - - # Multi-scale autocorrelation analysis - if len(sig) > 200: - scales = [50, 100, 200] - coherence_scores = [] - - for scale in scales: - if len(sig) > 2 * scale: - seg1 = sig[:scale] - seg2 = sig[scale:2*scale] - seg3 = sig[-scale:] - - # Cross-correlations between segments - corr12, _ = pearsonr(seg1, seg2) - corr13, _ = pearsonr(seg1, seg3) - corr23, _ = pearsonr(seg2, seg3) - - avg_corr = np.mean([abs(c) for c in [corr12, corr13, corr23] if not np.isnan(c)]) - coherence_scores.append(1.0 - avg_corr) - - return np.mean(coherence_scores) if coherence_scores else 0.5 - else: - return 0.0 - else: - # Multi-axis coherence analysis - coherence_loss = 0 - n_axes = data.shape[1] - pair_count = 0 - - for i in range(n_axes): - for j in range(i+1, n_axes): - try: - # Spectral coherence using scipy.signal.coherence - f, Cxy = coherence(data[:, i], data[:, j], fs=self.sample_rate, nperseg=min(1024, data.shape[0]//4)) - avg_coherence = np.mean(Cxy) - if not (np.isnan(avg_coherence) or np.isinf(avg_coherence)): - coherence_loss += (1.0 - avg_coherence) - pair_count += 1 - except: - # Fallback to simple correlation if coherence fails - try: - corr, _ = pearsonr(data[:, i], data[:, j]) - if not (np.isnan(corr) or np.isinf(corr)): - coherence_loss += (1.0 - abs(corr)) - pair_count += 1 - except: - pass - - # Normalize by number of valid pairs - return coherence_loss / pair_count if pair_count > 0 else 0.0 - - def detect_xi10_causal_violation(self, data): - """Enhanced temporal causality analysis""" - # For aerospace applications, this could detect synchronization issues - if len(data.shape) > 1 and data.shape[1] > 1: - # Cross-correlation delay analysis between channels - sig1 = data[:, 0] - sig2 = data[:, 1] - - try: - # Cross-correlation to find delays - correlation = np.correlate(sig1, sig2, mode='full') - delay = np.argmax(correlation) - len(sig2) + 1 - - # Normalize delay by signal length - relative_delay = abs(delay) / len(sig1) - - # Causality violation if delay is too large - return min(1.0, relative_delay * 10) - except: - # Fallback to simple correlation analysis - try: - corr, _ = pearsonr(sig1, sig2) - # Large correlation suggests possible causality issues - return min(1.0, abs(corr) * 0.5) if not (np.isnan(corr) or np.isinf(corr)) else 0.0 - except: - return 0.0 - else: - return 0.0 - - def compute_full_contradiction_analysis(self, data): - """Enhanced contradiction analysis with aerospace-grade metrics""" - start_time = time.time() - - xi = {} - xi[0] = self.detect_xi0_existential_collapse(data) - xi[1] = self.detect_xi1_boundary_overflow(data) - xi[2] = self.detect_xi2_role_conflict(data) - xi[3] = self.detect_xi3_symmetry_deadlock(data) - xi[4] = self.detect_xi4_temporal_instability(data) - xi[5] = self.detect_xi5_cycle_fracture(data) - xi[6] = self.detect_xi6_harmonic_asymmetry(data) - xi[7] = self.detect_xi7_curvature_overflow(data) - xi[8] = self.detect_xi8_emergence_boundary(data) - xi[9] = self.detect_xi9_longrange_coherence(data) - xi[10] = self.detect_xi10_causal_violation(data) - - # Enhanced metrics - phi = sum(self.weights[k] * xi[k] for k in xi.keys()) - health_score = 1.0 - xi[8] - computational_work = sum(self.weights[k] * xi[k] * self.computational_costs[k] for k in xi.keys()) - - # Processing time for real-time assessment - processing_time = time.time() - start_time - - # Enhanced rule-based classification - rule_fault = self.classify_fault_aerospace_grade(xi) - - # Confidence assessment - confidence = self.assess_classification_confidence(xi) - return { - 'xi': xi, - 'phi': phi, - 'health_score': health_score, - 'computational_work': computational_work, - 'processing_time': processing_time, - 'rule_fault': rule_fault, - 'confidence': confidence, - 'weights': self.weights + "phi": phi, + "w": w, + "z": z, + "original_count": n_points, + "final_count": len(phi), + "alpha": alpha_val, + "srl": srl_val } + + except Exception as e: + print(f"Error extracting CMT data from CSV row: {e}") + return None - def classify_fault_aerospace_grade(self, xi): - """Aerospace-grade fault classification with hierarchical logic""" - - # Critical faults (immediate attention) - if xi[0] > self.thresholds['xi0_critical']: - if xi[7] > 0.3: # High kurtosis + transients = bearing failure - return "critical_bearing_failure" - else: - return "critical_impact_damage" - - # Severe faults - if xi[7] > 0.4: # Very high kurtosis - return "severe_bearing_degradation" - - # Moderate faults - if xi[6] > self.thresholds['xi6_harmonic']: - if xi[6] > 0.2: # Strong harmonics - return "imbalance_severe" - elif xi[3] > 0.3: # With phase issues - return "misalignment_coupling" - else: - return "imbalance_moderate" - - # Early stage faults - if xi[8] > self.thresholds['xi8_emergence']: - if xi[5] > 0.3: # Spectral changes - return "incipient_bearing_wear" - elif xi[9] > 0.4: # Coherence loss - return "structural_loosening" - else: - return "unknown_degradation" - - # Sensor/instrumentation issues - if xi[1] > 0.1 or xi[4] > 0.2: - return "sensor_instrumentation_fault" - - # System healthy - if xi[8] < 0.05: - return "healthy" - else: - return "monitoring_required" - - def assess_classification_confidence(self, xi): - """Assess confidence in fault classification""" - - # High confidence indicators - high_confidence_conditions = [ - xi[0] > 0.01, # Clear transients - xi[6] > 0.15, # Strong harmonics - xi[7] > 0.3, # High kurtosis - xi[8] < 0.02 or xi[8] > 0.3 # Very healthy or clearly degraded - ] - - confidence = 0.5 # Base confidence - - # Increase confidence for clear indicators - for condition in high_confidence_conditions: - if condition: - confidence += 0.1 - - # Decrease confidence for ambiguous cases - if 0.05 < xi[8] < 0.15: # Borderline emergence - confidence -= 0.2 - - return min(1.0, max(0.0, confidence)) - -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π NASA-GRADE SIGNAL SIMULATOR -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - -class NASAGradeSimulator: - """ - Ultra-realistic simulation of aerospace-grade machinery vibrations - with multi-modal noise, environmental effects, and complex failure modes. - """ - - @staticmethod - def generate_aerospace_vibration(fault_type, length=16384, sample_rate=100000, - rpm=6000, base_noise=0.02, environmental_factor=1.0, - thermal_noise=True, emi_noise=True, - sensor_degradation=0.0, load_variation=True): - """Generate ultra-realistic aerospace vibration with complex environmental effects""" - - t = np.linspace(0, length/sample_rate, length) - fundamental = rpm / 60.0 # Hz - - # === MULTI-MODAL NOISE GENERATION === - - # 1. Base mechanical noise - mechanical_noise = np.random.normal(0, base_noise, (length, 3)) - - # 2. Thermal noise (temperature-dependent) - if thermal_noise: - thermal_drift = 0.01 * environmental_factor * np.sin(2*np.pi*0.05*t) # 0.05 Hz thermal cycle - thermal_noise_amp = base_noise * 0.3 * environmental_factor - thermal_component = np.random.normal(0, thermal_noise_amp, (length, 3)) - thermal_component += np.column_stack([thermal_drift, thermal_drift*0.8, thermal_drift*1.2]) - else: - thermal_component = np.zeros((length, 3)) - - # 3. Electromagnetic interference (EMI) - if emi_noise: - # Power line interference (50/60 Hz and harmonics) - power_freq = 60.0 # Hz - emi_signal = np.zeros(length) - for harmonic in [1, 2, 3, 5]: # Typical EMI harmonics - emi_signal += 0.005 * environmental_factor * np.sin(2*np.pi*power_freq*harmonic*t + np.random.uniform(0, 2*np.pi)) - - # Random EMI spikes - n_spikes = int(environmental_factor * np.random.poisson(3)) - for _ in range(n_spikes): - spike_time = np.random.uniform(0, t[-1]) - spike_idx = int(spike_time * sample_rate) - if spike_idx < length: - spike_duration = int(0.001 * sample_rate) # 1ms spikes - end_idx = min(spike_idx + spike_duration, length) - emi_signal[spike_idx:end_idx] += np.random.uniform(0.01, 0.05) * environmental_factor - - emi_component = np.column_stack([emi_signal, emi_signal*0.6, emi_signal*0.4]) - else: - emi_component = np.zeros((length, 3)) - - # 4. Load variation effects - if load_variation: - load_frequency = 0.1 # Hz - slow load variations - load_amplitude = 0.2 * environmental_factor - load_modulation = 1.0 + load_amplitude * np.sin(2*np.pi*load_frequency*t) - else: - load_modulation = np.ones(length) - - # === FAULT SIGNATURE GENERATION === - - def generate_aerospace_fault(fault): - """Generate aerospace-specific fault signatures""" - - if fault == "healthy": - return np.zeros((length, 3)) - - elif fault == "rotor_imbalance": - # High-precision rotor imbalance with load modulation - sig = 0.3 * np.sin(2*np.pi*fundamental*t) * load_modulation - # Add slight asymmetry between axes - return np.column_stack([sig, 0.85*sig, 1.1*sig]) - - elif fault == "shaft_misalignment": - # Complex misalignment with multiple harmonics - sig2 = 0.25 * np.sin(2*np.pi*2*fundamental*t + np.pi/4) - sig3 = 0.15 * np.sin(2*np.pi*3*fundamental*t + np.pi/3) - sig4 = 0.10 * np.sin(2*np.pi*4*fundamental*t + np.pi/6) - sig = (sig2 + sig3 + sig4) * load_modulation - return np.column_stack([sig, 1.2*sig, 0.9*sig]) - - elif fault == "bearing_outer_race": - # Precise bearing outer race defect - bpfo = fundamental * 3.585 # Typical outer race passing frequency - envelope_freq = fundamental # Modulation by shaft rotation - - # Generate impulse train - impulse_times = np.arange(0, t[-1], 1/bpfo) - sig = np.zeros(length) - - for imp_time in impulse_times: - idx = int(imp_time * sample_rate) - if idx < length: - # Each impulse is a damped oscillation - impulse_duration = int(0.002 * sample_rate) # 2ms impulse - end_idx = min(idx + impulse_duration, length) - impulse_t = np.arange(end_idx - idx) / sample_rate - - # Damped sinusoid representing bearing resonance - resonance_freq = 5000 # Hz - typical bearing resonance - damping = 1000 # Damping coefficient - impulse = np.exp(-damping * impulse_t) * np.sin(2*np.pi*resonance_freq*impulse_t) - - # Amplitude modulation by envelope frequency - amplitude = 0.8 * (1 + 0.5*np.sin(2*np.pi*envelope_freq*imp_time)) - sig[idx:end_idx] += amplitude * impulse - - return np.column_stack([sig, 0.7*sig, 0.9*sig]) - - elif fault == "bearing_inner_race": - # Inner race defect with higher frequency - bpfi = fundamental * 5.415 - - impulse_times = np.arange(0, t[-1], 1/bpfi) - sig = np.zeros(length) - - for imp_time in impulse_times: - idx = int(imp_time * sample_rate) - if idx < length: - impulse_duration = int(0.0015 * sample_rate) # Shorter impulses - end_idx = min(idx + impulse_duration, length) - impulse_t = np.arange(end_idx - idx) / sample_rate - - resonance_freq = 6000 # Slightly higher resonance - damping = 1200 - impulse = np.exp(-damping * impulse_t) * np.sin(2*np.pi*resonance_freq*impulse_t) - - amplitude = 0.6 * np.random.uniform(0.8, 1.2) # More random amplitude - sig[idx:end_idx] += amplitude * impulse - - return np.column_stack([sig, 0.8*sig, 0.6*sig]) - - elif fault == "gear_tooth_defect": - # Single tooth defect in gear mesh - gear_teeth = 24 # Number of teeth - gmf = fundamental * gear_teeth # Gear mesh frequency - - # Base gear mesh signal - gmf_signal = 0.2 * np.sin(2*np.pi*gmf*t) - - # Defect once per revolution - defect_times = np.arange(0, t[-1], 1/fundamental) - defect_signal = np.zeros(length) - - for def_time in defect_times: - idx = int(def_time * sample_rate) - if idx < length: - # Sharp impact from defective tooth - impact_duration = int(0.0005 * sample_rate) # 0.5ms impact - end_idx = min(idx + impact_duration, length) - impact_t = np.arange(end_idx - idx) / sample_rate - - # High-frequency impact with multiple resonances - impact = 0.0 - for res_freq in [8000, 12000, 16000]: # Multiple resonances - impact += np.exp(-2000 * impact_t) * np.sin(2*np.pi*res_freq*impact_t) - - defect_signal[idx:end_idx] += 1.5 * impact - - total_signal = gmf_signal + defect_signal - return np.column_stack([total_signal, 0.9*total_signal, 0.8*total_signal]) - - elif fault == "turbine_blade_crack": - # Aerospace-specific: turbine blade natural frequency excitation - blade_freq = 1200 # Hz - typical turbine blade natural frequency - - # Crack causes modulation of blade response - crack_modulation = 0.1 * np.sin(2*np.pi*fundamental*t) # Once per revolution modulation - blade_response = 0.15 * (1 + crack_modulation) * np.sin(2*np.pi*blade_freq*t) - - # Add random amplitude variation due to crack growth - random_variation = 0.05 * np.random.normal(0, 1, length) - blade_response += random_variation - - return np.column_stack([blade_response, 0.3*blade_response, 0.2*blade_response]) - - elif fault == "seal_degradation": - # Aerospace seal degradation - creates aerodynamic noise - # Multiple frequency components from turbulent flow - flow_noise = np.zeros(length) - - # Broadband noise with specific frequency peaks - for freq in np.random.uniform(200, 2000, 10): # Random aerodynamic frequencies - amplitude = 0.05 * np.random.uniform(0.5, 1.5) - flow_noise += amplitude * np.sin(2*np.pi*freq*t + np.random.uniform(0, 2*np.pi)) - - # Modulation by operating frequency - flow_noise *= (1 + 0.3*np.sin(2*np.pi*fundamental*t)) - - return np.column_stack([flow_noise, 1.2*flow_noise, 0.8*flow_noise]) - - elif fault == "sensor_degradation": - # Realistic sensor degradation effects - sig = np.zeros(length) - - # Gradual bias drift - bias_drift = 0.5 * environmental_factor * t / t[-1] - - # Random spikes from connector issues - n_spikes = int(environmental_factor * np.random.poisson(2)) - for _ in range(n_spikes): - spike_idx = np.random.randint(length) - spike_amplitude = np.random.uniform(2.0, 8.0) * environmental_factor - spike_duration = np.random.randint(1, 10) - end_idx = min(spike_idx + spike_duration, length) - sig[spike_idx:end_idx] = spike_amplitude - - # Frequency response degradation (high-freq rolloff) - from scipy.signal import butter, filtfilt - if environmental_factor > 1.5: # Severe degradation - nyquist = sample_rate / 2 - cutoff_freq = 5000 # Hz - sensor bandwidth reduction - b, a = butter(2, cutoff_freq / nyquist, btype='low') - sig = filtfilt(b, a, sig) - - sig += bias_drift - return np.column_stack([sig, 0.1*sig, 0.1*sig]) - - else: - return np.zeros((length, 3)) - - # Handle compound faults - if "compound" in fault_type: - components = fault_type.replace("compound_", "").split("_") - combined_sig = np.zeros((length, 3)) - - for i, component in enumerate(components): - component_sig = generate_aerospace_fault(component) - # Reduce amplitude for each additional component - amplitude_factor = 0.8 ** i - combined_sig += amplitude_factor * component_sig - - fault_signal = combined_sig - else: - fault_signal = generate_aerospace_fault(fault_type) - - # === COMBINE ALL COMPONENTS === - - base_signal = mechanical_noise + thermal_component + emi_component - total_signal = base_signal + fault_signal - - # === SENSOR DEGRADATION SIMULATION === - - if sensor_degradation > 0: - # Simulate various sensor degradation effects - - # 1. Sensitivity degradation - sensitivity_loss = 1.0 - sensor_degradation * 0.3 - total_signal *= sensitivity_loss - - # 2. Noise floor increase - degraded_noise = np.random.normal(0, base_noise * sensor_degradation, (length, 3)) - total_signal += degraded_noise - - # 3. Frequency response degradation - if sensor_degradation > 0.5: - from scipy.signal import butter, filtfilt - nyquist = sample_rate / 2 - cutoff_freq = 20000 * (1 - sensor_degradation) # Bandwidth reduction - b, a = butter(3, cutoff_freq / nyquist, btype='low') - for axis in range(3): - total_signal[:, axis] = filtfilt(b, a, total_signal[:, axis]) - - # === REALISTIC DATA CORRUPTION === - - corruption_probability = 0.1 * environmental_factor - if np.random.random() < corruption_probability: - corruption_type = np.random.choice(['dropout', 'saturation', 'aliasing', 'sync_loss'], - p=[0.3, 0.3, 0.2, 0.2]) - - if corruption_type == 'dropout': - # Communication dropout - dropout_duration = int(np.random.uniform(0.001, 0.01) * sample_rate) # 1-10ms - dropout_start = np.random.randint(0, length - dropout_duration) - total_signal[dropout_start:dropout_start+dropout_duration, :] = 0 - - elif corruption_type == 'saturation': - # ADC saturation - saturation_level = np.random.uniform(3.0, 6.0) - total_signal = np.clip(total_signal, -saturation_level, saturation_level) - - elif corruption_type == 'aliasing': - # Sample rate mismatch aliasing - downsample_factor = np.random.randint(2, 4) - downsampled = total_signal[::downsample_factor, :] - - # Interpolate back to original length - old_indices = np.arange(0, length, downsample_factor) - new_indices = np.arange(length) - - for axis in range(3): - if len(old_indices) > 1: - f_interp = interpolate.interp1d(old_indices, downsampled[:, axis], - kind='linear', fill_value='extrapolate') - total_signal[:, axis] = f_interp(new_indices) - - elif corruption_type == 'sync_loss': - # Synchronization loss between axes - if total_signal.shape[1] > 1: - sync_offset = np.random.randint(1, 50) # Sample offset - total_signal[:, 1] = np.roll(total_signal[:, 1], sync_offset) - if total_signal.shape[1] > 2: - sync_offset = np.random.randint(1, 50) - total_signal[:, 2] = np.roll(total_signal[:, 2], -sync_offset) - - return total_signal - -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π¬ STATE-OF-THE-ART COMPETITOR METHODS -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - -class StateOfTheArtCompetitors: - """Implementation of current best-practice methods in fault detection""" - - @staticmethod - def wavelet_classifier(samples, sample_rate=100000): - """Advanced wavelet-based fault detection with fallback""" - predictions = [] - - for sample in samples: - sig = sample[:, 0] if len(sample.shape) > 1 else sample - - if HAS_PYWAVELETS: - try: - # Multi-resolution wavelet decomposition - coeffs = pywt.wavedec(sig, 'db8', level=6) - - # Energy distribution across scales - energies = [np.sum(c**2) for c in coeffs] - total_energy = sum(energies) - energy_ratios = [e/total_energy for e in energies] if total_energy > 0 else [0]*len(energies) - - # Decision logic based on energy distribution - if energy_ratios[0] > 0.6: # High energy in approximation (low freq) - predictions.append("rotor_imbalance") - elif energy_ratios[1] > 0.3: # High energy in detail level 1 - predictions.append("bearing_outer_race") - elif energy_ratios[2] > 0.25: # High energy in detail level 2 - predictions.append("bearing_inner_race") - elif max(energy_ratios[3:]) > 0.2: # High energy in higher details - predictions.append("gear_tooth_defect") - else: - predictions.append("healthy") - - except Exception: - # Fallback to frequency band analysis - predictions.append(StateOfTheArtCompetitors._frequency_band_classifier(sig, sample_rate)) - else: - # Fallback to frequency band analysis when PyWavelets not available - predictions.append(StateOfTheArtCompetitors._frequency_band_classifier(sig, sample_rate)) - - return predictions - - @staticmethod - def _frequency_band_classifier(sig, sample_rate): - """Fallback frequency band analysis when wavelets not available""" - f, Pxx = welch(sig, fs=sample_rate, nperseg=1024) - - # Define frequency bands - low_freq = np.sum(Pxx[f < 100]) # 0-100 Hz - mid_freq = np.sum(Pxx[(f >= 100) & (f < 1000)]) # 100-1000 Hz - high_freq = np.sum(Pxx[f >= 1000]) # >1000 Hz - total_energy = np.sum(Pxx) - - if total_energy > 0: - low_ratio = low_freq / total_energy - mid_ratio = mid_freq / total_energy - high_ratio = high_freq / total_energy - - if low_ratio > 0.6: - return "rotor_imbalance" - elif mid_ratio > 0.4: - return "bearing_outer_race" - elif high_ratio > 0.3: - return "bearing_inner_race" - else: - return "gear_tooth_defect" - else: - return "healthy" - - @staticmethod - def envelope_analysis_classifier(samples, sample_rate=100000): - """Industry-standard envelope analysis for bearing fault detection""" - predictions = [] - - for sample in samples: - sig = sample[:, 0] if len(sample.shape) > 1 else sample - - # Envelope analysis using Hilbert transform - analytic_signal = hilbert(sig) - envelope = np.abs(analytic_signal) - - # Spectral analysis of envelope - f_env, Pxx_env = welch(envelope, fs=sample_rate, nperseg=1024) - - # Look for bearing fault frequencies in envelope spectrum - # Assuming typical bearing frequencies - bpfo_freq = 60 # Outer race frequency - bpfi_freq = 90 # Inner race frequency - - # Find peaks in envelope spectrum - peaks, _ = find_peaks(Pxx_env, height=np.max(Pxx_env)*0.1) - peak_freqs = f_env[peaks] - - # Classification based on detected frequencies - if any(abs(pf - bpfo_freq) < 5 for pf in peak_freqs): - predictions.append("bearing_outer_race") - elif any(abs(pf - bpfi_freq) < 5 for pf in peak_freqs): - predictions.append("bearing_inner_race") - elif kurtosis(envelope) > 4: - predictions.append("bearing_outer_race") # High kurtosis indicates impacts - elif np.std(envelope) > 0.5: - predictions.append("imbalance") - else: - predictions.append("healthy") - - return predictions - - @staticmethod - def spectral_kurtosis_classifier(samples, sample_rate=100000): - """Advanced spectral kurtosis method for fault detection""" - predictions = [] - - for sample in samples: - sig = sample[:, 0] if len(sample.shape) > 1 else sample - - # Compute spectrogram - f, t_spec, Sxx = spectrogram(sig, fs=sample_rate, nperseg=512, noverlap=256) - - # Compute kurtosis across time for each frequency - spectral_kurt = [] - for freq_idx in range(len(f)): - freq_time_series = Sxx[freq_idx, :] - if len(freq_time_series) > 3: # Need at least 4 points for kurtosis - kurt_val = kurtosis(freq_time_series) - spectral_kurt.append(kurt_val) - else: - spectral_kurt.append(0) - - spectral_kurt = np.array(spectral_kurt) - - # Find frequency bands with high kurtosis - high_kurt_mask = spectral_kurt > 3 - high_kurt_freqs = f[high_kurt_mask] - - # Classification based on frequency ranges with high kurtosis - if any((1000 <= freq <= 5000) for freq in high_kurt_freqs): - predictions.append("bearing_outer_race") - elif any((5000 <= freq <= 15000) for freq in high_kurt_freqs): - predictions.append("bearing_inner_race") - elif any((500 <= freq <= 1000) for freq in high_kurt_freqs): - predictions.append("gear_tooth_defect") - elif np.max(spectral_kurt) > 2: - predictions.append("imbalance") - else: - predictions.append("healthy") - - return predictions - - @staticmethod - def deep_learning_classifier(samples, labels_train=None, samples_train=None): - """Deep learning baseline using CNN""" - if not HAS_TENSORFLOW: - # Fallback to simple classification if TensorFlow not available - return ["healthy"] * len(samples) - - # Prepare data for CNN - def prepare_spectrogram_data(samples_list): - spectrograms = [] - for sample in samples_list: - sig = sample[:, 0] if len(sample.shape) > 1 else sample - f, t, Sxx = spectrogram(sig, fs=100000, nperseg=256, noverlap=128) - Sxx_log = np.log10(Sxx + 1e-12) # Log scale - # Resize to fixed shape - if Sxx_log.shape != (129, 63): # Expected shape from spectrogram - # Pad or truncate to standard size - target_shape = (64, 64) # Square for CNN - Sxx_resized = np.zeros(target_shape) - min_freq = min(Sxx_log.shape[0], target_shape[0]) - min_time = min(Sxx_log.shape[1], target_shape[1]) - Sxx_resized[:min_freq, :min_time] = Sxx_log[:min_freq, :min_time] - spectrograms.append(Sxx_resized) - else: - # Resize to 64x64 - from scipy.ndimage import zoom - zoom_factors = (64/Sxx_log.shape[0], 64/Sxx_log.shape[1]) - Sxx_resized = zoom(Sxx_log, zoom_factors) - spectrograms.append(Sxx_resized) - - return np.array(spectrograms) - - # If training data provided, train a simple CNN - if samples_train is not None and labels_train is not None: - try: - # Prepare training data - X_train_spec = prepare_spectrogram_data(samples_train) - X_train_spec = X_train_spec.reshape(-1, 64, 64, 1) - - # Encode labels - unique_labels = np.unique(labels_train) - label_to_int = {label: i for i, label in enumerate(unique_labels)} - y_train_int = np.array([label_to_int[label] for label in labels_train]) - y_train_cat = tf.keras.utils.to_categorical(y_train_int, len(unique_labels)) - - # Simple CNN model - model = tf.keras.Sequential([ - tf.keras.layers.Conv2D(32, (3, 3), activation='relu', input_shape=(64, 64, 1)), - tf.keras.layers.MaxPooling2D((2, 2)), - tf.keras.layers.Conv2D(64, (3, 3), activation='relu'), - tf.keras.layers.MaxPooling2D((2, 2)), - tf.keras.layers.Flatten(), - tf.keras.layers.Dense(128, activation='relu'), - tf.keras.layers.Dropout(0.5), - tf.keras.layers.Dense(len(unique_labels), activation='softmax') - ]) - - model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) - - # Train model (limited epochs for demo) - model.fit(X_train_spec, y_train_cat, epochs=5, batch_size=32, verbose=0) - - # Prepare test data and predict - X_test_spec = prepare_spectrogram_data(samples) - X_test_spec = X_test_spec.reshape(-1, 64, 64, 1) - - predictions_int = model.predict(X_test_spec, verbose=0) - predictions_labels = [unique_labels[np.argmax(pred)] for pred in predictions_int] - - return predictions_labels - - except Exception as e: - print(f"Deep learning classifier failed: {e}") - # Fallback to simple rule-based - return ["healthy"] * len(samples) - else: - # No training data provided - return ["healthy"] * len(samples) - -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π NASA-GRADE FLAGSHIP DEMONSTRATION -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ +def generate_holographic_field(z: np.ndarray, phi: np.ndarray, resolution: int): + if z is None or phi is None or len(z) < 4: return None -def run_nasa_grade_demonstration(): - """ - π NASA-GRADE FLAGSHIP DEMONSTRATION + points = np.vstack([np.real(z), np.imag(z)]).T + grid_x, grid_y = np.mgrid[ + np.min(points[:,0]):np.max(points[:,0]):complex(0, resolution), + np.min(points[:,1]):np.max(points[:,1]):complex(0, resolution) + ] - Ultra-realistic validation under aerospace conditions with statistical rigor - """ + grid_phi_real = griddata(points, np.real(phi), (grid_x, grid_y), method='cubic') + grid_phi_imag = griddata(points, np.imag(phi), (grid_x, grid_y), method='cubic') - print(""" - π― INITIALIZING NASA-GRADE DEMONSTRATION - ======================================= - β’ 9 aerospace-relevant fault types + compound failures - β’ 600+ samples with extreme environmental conditions - β’ State-of-the-art competitor methods (wavelets, envelope analysis, deep learning) - β’ Statistical significance testing with confidence intervals - β’ Early detection capability analysis - β’ Real-time performance validation - """) - - # Enhanced fault types for aerospace applications - fault_types = [ - "healthy", - "rotor_imbalance", - "shaft_misalignment", - "bearing_outer_race", - "bearing_inner_race", - "gear_tooth_defect", - "turbine_blade_crack", - "seal_degradation", - "sensor_degradation", - "compound_imbalance_bearing", - "compound_misalignment_gear" - ] + grid_phi = np.nan_to_num(grid_phi_real + 1j * grid_phi_imag) - # Initialize NASA-grade CMT engine - engine = CMT_Vibration_Engine_NASA(sample_rate=100000, rpm=6000) - - # βββ STEP 1: ESTABLISH BASELINE βββ - print("π§ Establishing aerospace-grade baseline...") - healthy_samples = [] - for i in range(10): # More baseline samples for robustness - healthy_data = NASAGradeSimulator.generate_aerospace_vibration( - "healthy", - length=16384, - sample_rate=100000, - rpm=6000, - base_noise=0.01, # Very low noise for pristine baseline - environmental_factor=0.5, # Controlled environment - thermal_noise=False, - emi_noise=False, - sensor_degradation=0.0 - ) - healthy_samples.append(healthy_data) - - baseline_data = np.mean(healthy_samples, axis=0) - engine.establish_baseline(baseline_data) - print("β Aerospace baseline established") - - # βββ STEP 2: GENERATE EXTREME CONDITION DATASET βββ - print("π Generating NASA-grade test dataset...") - - samples_per_fault = 55 # Total: 605 samples - all_samples = [] - all_labels = [] - all_srl_features = [] - all_processing_times = [] - - # Extreme condition parameters - rpms = [3000, 4500, 6000, 7500, 9000] # Wide RPM range - noise_levels = [0.02, 0.05, 0.08, 0.12, 0.15] # From pristine to very noisy - environmental_factors = [1.0, 1.5, 2.0, 2.5, 3.0] # Extreme environmental conditions - sensor_degradations = [0.0, 0.1, 0.3, 0.5, 0.7] # From perfect to severely degraded sensors - - print(" Testing conditions:") - print(f" β’ RPM range: {min(rpms)} - {max(rpms)} RPM") - print(f" β’ Noise levels: {min(noise_levels):.3f} - {max(noise_levels):.3f}") - print(f" β’ Environmental factors: {min(environmental_factors)} - {max(environmental_factors)}x") - print(f" β’ Sensor degradation: {min(sensor_degradations):.1%} - {max(sensor_degradations):.1%}") - - for fault_type in fault_types: - print(f" Generating {fault_type} samples...") - for i in range(samples_per_fault): - # Extreme condition sampling - rpm = np.random.choice(rpms) - noise = np.random.choice(noise_levels) - env_factor = np.random.choice(environmental_factors) - sensor_deg = np.random.choice(sensor_degradations) - - # Update engine parameters - engine.rpm = rpm - - # Generate sample under extreme conditions - sample = NASAGradeSimulator.generate_aerospace_vibration( - fault_type, - length=16384, - sample_rate=100000, - rpm=rpm, - base_noise=noise, - environmental_factor=env_factor, - thermal_noise=True, - emi_noise=True, - sensor_degradation=sensor_deg, - load_variation=True - ) + return grid_x, grid_y, grid_phi - # SRL-SEFA analysis - analysis = engine.compute_full_contradiction_analysis(sample) +def create_holography_plot(z, phi, resolution, wavelength): + field_data = generate_holographic_field(z, phi, resolution) + if field_data is None: return go.Figure(layout={"title": "Not enough data for holography"}) - # Store results - all_samples.append(sample) - all_labels.append(fault_type) - all_processing_times.append(analysis['processing_time']) + grid_x, grid_y, grid_phi = field_data + mag_phi = np.abs(grid_phi) + phase_phi = np.angle(grid_phi) - # Extended feature vector - feature_vector = ( - [analysis['xi'][k] for k in range(11)] + - [analysis['phi'], analysis['health_score'], analysis['computational_work'], - analysis['confidence']] - ) - all_srl_features.append(feature_vector) + # --- Wavelength to Colorscale Mapping --- + def wavelength_to_rgb(wl): + # Simple approximation to map visible spectrum to RGB + if 380 <= wl < 440: return f'rgb({-(wl - 440) / (440 - 380) * 255}, 0, 255)' # Violet + elif 440 <= wl < 495: return f'rgb(0, {(wl - 440) / (495 - 440) * 255}, 255)' # Blue + elif 495 <= wl < 570: return f'rgb(0, 255, {-(wl - 570) / (570 - 495) * 255})' # Green + elif 570 <= wl < 590: return f'rgb({(wl - 570) / (590 - 570) * 255}, 255, 0)' # Yellow + elif 590 <= wl < 620: return f'rgb(255, {-(wl - 620) / (620 - 590) * 255}, 0)' # Orange + elif 620 <= wl <= 750: return f'rgb(255, 0, 0)' # Red + return 'rgb(255,255,255)' + + mid_color = wavelength_to_rgb(wavelength) + custom_colorscale = [[0, 'rgb(20,0,40)'], [0.5, mid_color], [1, 'rgb(255,255,255)']] + + + fig = go.Figure() + # 1. The Holographic Surface (Topology + Phase Interference) + fig.add_trace(go.Surface( + x=grid_x, y=grid_y, z=mag_phi, + surfacecolor=phase_phi, + colorscale=custom_colorscale, + cmin=-np.pi, cmax=np.pi, + colorbar=dict(title='Ξ¦ Phase'), + name='Holographic Field', + contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True, highlightwidth=10) + )) + # 2. The original data points projected onto the surface + fig.add_trace(go.Scatter3d( + x=np.real(z), y=np.imag(z), z=np.abs(phi) + 0.05, # slight offset + mode='markers', + marker=dict(size=3, color='black', symbol='x'), + name='Data Points' + )) + # 3. The Vector Flow Field (using cones for direction) + grad_y, grad_x = np.gradient(mag_phi) + fig.add_trace(go.Cone( + x=grid_x.flatten(), y=grid_y.flatten(), z=mag_phi.flatten(), + u=-grad_x.flatten(), v=-grad_y.flatten(), w=np.full_like(mag_phi.flatten(), -0.1), + sizemode="absolute", sizeref=0.1, + anchor="tip", + colorscale='Greys', + showscale=False, + name='Vector Flow' + )) + fig.update_layout( + title="Interactive Holographic Field Reconstruction", + scene=dict( + xaxis_title="Re(z) - Encoded Signal", + yaxis_title="Im(z) - Encoded Signal", + zaxis_title="|Ξ¦| - Field Magnitude" + ), + margin=dict(l=0, r=0, b=0, t=40) + ) + return fig + +def create_diagnostic_plots(z, w): + """Creates a 2D plot showing the Aperture (z) and Lens Response (w).""" + if z is None or w is None: + return go.Figure(layout={"title": "Not enough data for diagnostic plots"}) + + fig = go.Figure() + + # Aperture (Encoded Signal) + fig.add_trace(go.Scatter( + x=np.real(z), y=np.imag(z), mode='markers', + marker=dict(size=5, color='blue', opacity=0.6), + name='Aperture (z)' + )) + + # Lens Response + fig.add_trace(go.Scatter( + x=np.real(w), y=np.imag(w), mode='markers', + marker=dict(size=5, color='red', opacity=0.6, symbol='x'), + name='Lens Response (w)' + )) + + fig.update_layout( + title="Diagnostic View: Aperture and Lens Response", + xaxis_title="Real Part", + yaxis_title="Imaginary Part", + legend_title="Signal Stage", + margin=dict(l=20, r=20, t=60, b=20) + ) + return fig - # Convert to arrays - X_srl = np.array(all_srl_features) - y = np.array(all_labels) - raw_samples = np.array(all_samples) - processing_times = np.array(all_processing_times) +def create_dual_holography_plot(z1, phi1, z2, phi2, resolution, wavelength, title1="Primary", title2="Comparison"): + """Creates side-by-side holographic visualizations for comparison.""" + field_data1 = generate_holographic_field(z1, phi1, resolution) + field_data2 = generate_holographic_field(z2, phi2, resolution) + + if field_data1 is None or field_data2 is None: + return go.Figure(layout={"title": "Insufficient data for dual holography"}) - print(f"β Extreme conditions dataset: {len(X_srl)} samples, {len(fault_types)} fault types") - print(f" Average processing time: {np.mean(processing_times)*1000:.2f}ms") + grid_x1, grid_y1, grid_phi1 = field_data1 + grid_x2, grid_y2, grid_phi2 = field_data2 + + mag_phi1, phase_phi1 = np.abs(grid_phi1), np.angle(grid_phi1) + mag_phi2, phase_phi2 = np.abs(grid_phi2), np.angle(grid_phi2) + + # Wavelength to colorscale mapping + def wavelength_to_rgb(wl): + if 380 <= wl < 440: return f'rgb({int(-(wl - 440) / (440 - 380) * 255)}, 0, 255)' + elif 440 <= wl < 495: return f'rgb(0, {int((wl - 440) / (495 - 440) * 255)}, 255)' + elif 495 <= wl < 570: return f'rgb(0, 255, {int(-(wl - 570) / (570 - 495) * 255)})' + elif 570 <= wl < 590: return f'rgb({int((wl - 570) / (590 - 570) * 255)}, 255, 0)' + elif 590 <= wl < 620: return f'rgb(255, {int(-(wl - 620) / (620 - 590) * 255)}, 0)' + elif 620 <= wl <= 750: return 'rgb(255, 0, 0)' + return 'rgb(255,255,255)' + + mid_color = wavelength_to_rgb(wavelength) + custom_colorscale = [[0, 'rgb(20,0,40)'], [0.5, mid_color], [1, 'rgb(255,255,255)']] - # βββ STEP 3: TRAIN-TEST SPLIT βββ - X_train, X_test, y_train, y_test, samples_train, samples_test = train_test_split( - X_srl, y, raw_samples, test_size=0.25, stratify=y, random_state=42 + fig = make_subplots( + rows=1, cols=2, + specs=[[{'type': 'scene'}, {'type': 'scene'}]], + subplot_titles=[title1, title2] ) - # Ensure labels are numpy arrays - y_train = np.array(y_train) - y_test = np.array(y_test) - - # βββ STEP 4: IMPLEMENT STATE-OF-THE-ART COMPETITORS βββ - print("π Implementing state-of-the-art competitors...") + # Left plot (Primary) + fig.add_trace(go.Surface( + x=grid_x1, y=grid_y1, z=mag_phi1, + surfacecolor=phase_phi1, + colorscale=custom_colorscale, + cmin=-np.pi, cmax=np.pi, + showscale=False, + name=title1, + contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True) + ), row=1, col=1) + + # Right plot (Comparison) + fig.add_trace(go.Surface( + x=grid_x2, y=grid_y2, z=mag_phi2, + surfacecolor=phase_phi2, + colorscale=custom_colorscale, + cmin=-np.pi, cmax=np.pi, + showscale=False, + name=title2, + contours_z=dict(show=True, usecolormap=True, highlightcolor="limegreen", project_z=True) + ), row=1, col=2) + + # Add data points + if z1 is not None and phi1 is not None: + fig.add_trace(go.Scatter3d( + x=np.real(z1), y=np.imag(z1), z=np.abs(phi1) + 0.05, + mode='markers', marker=dict(size=3, color='black', symbol='x'), + name=f'{title1} Points', showlegend=False + ), row=1, col=1) + + if z2 is not None and phi2 is not None: + fig.add_trace(go.Scatter3d( + x=np.real(z2), y=np.imag(z2), z=np.abs(phi2) + 0.05, + mode='markers', marker=dict(size=3, color='black', symbol='x'), + name=f'{title2} Points', showlegend=False + ), row=1, col=2) + + fig.update_layout( + title="Side-by-Side Cross-Species Holographic Comparison", + scene=dict( + xaxis_title="Re(z)", yaxis_title="Im(z)", zaxis_title="|Ξ¦|", + camera=dict(eye=dict(x=1.5, y=1.5, z=1.5)) + ), + scene2=dict( + xaxis_title="Re(z)", yaxis_title="Im(z)", zaxis_title="|Ξ¦|", + camera=dict(eye=dict(x=1.5, y=1.5, z=1.5)) + ), + margin=dict(l=0, r=0, b=0, t=60), + height=600 + ) + return fig - competitors = StateOfTheArtCompetitors() +def create_dual_diagnostic_plots(z1, w1, z2, w2, title1="Primary", title2="Comparison"): + """Creates side-by-side diagnostic plots for cross-species comparison.""" + fig = make_subplots( + rows=1, cols=2, + subplot_titles=[f"{title1}: Aperture & Lens Response", f"{title2}: Aperture & Lens Response"] + ) - # Get competitor predictions - print(" β’ Wavelet-based classification...") - y_pred_wavelet = competitors.wavelet_classifier(samples_test) + if z1 is not None and w1 is not None: + # Primary aperture and response + fig.add_trace(go.Scatter( + x=np.real(z1), y=np.imag(z1), mode='markers', + marker=dict(size=5, color='blue', opacity=0.6), + name=f'{title1} Aperture', showlegend=True + ), row=1, col=1) + + fig.add_trace(go.Scatter( + x=np.real(w1), y=np.imag(w1), mode='markers', + marker=dict(size=5, color='red', opacity=0.6, symbol='x'), + name=f'{title1} Response', showlegend=True + ), row=1, col=1) + + if z2 is not None and w2 is not None: + # Comparison aperture and response + fig.add_trace(go.Scatter( + x=np.real(z2), y=np.imag(z2), mode='markers', + marker=dict(size=5, color='darkblue', opacity=0.6), + name=f'{title2} Aperture', showlegend=True + ), row=1, col=2) + + fig.add_trace(go.Scatter( + x=np.real(w2), y=np.imag(w2), mode='markers', + marker=dict(size=5, color='darkred', opacity=0.6, symbol='x'), + name=f'{title2} Response', showlegend=True + ), row=1, col=2) + + fig.update_layout( + title="Cross-Species Diagnostic Comparison", + height=400, + margin=dict(l=20, r=20, t=60, b=20) + ) + fig.update_xaxes(title_text="Real Part", row=1, col=1) + fig.update_yaxes(title_text="Imaginary Part", row=1, col=1) + fig.update_xaxes(title_text="Real Part", row=1, col=2) + fig.update_yaxes(title_text="Imaginary Part", row=1, col=2) + + return fig - print(" β’ Envelope analysis classification...") - y_pred_envelope = competitors.envelope_analysis_classifier(samples_test) - print(" β’ Spectral kurtosis classification...") - y_pred_spectral_kurt = competitors.spectral_kurtosis_classifier(samples_test) - - print(" β’ Deep learning classification...") - y_pred_deep = competitors.deep_learning_classifier(samples_test, y_train, samples_train) - - # βββ STEP 5: SRL-SEFA + ADVANCED ML βββ - print("π§ Training SRL-SEFA + Advanced ML ensemble...") - - # Scale features - scaler = StandardScaler() - X_train_scaled = scaler.fit_transform(X_train) - X_test_scaled = scaler.transform(X_test) - - # Multiple ML models for ensemble - rf_classifier = RandomForestClassifier(n_estimators=300, max_depth=20, random_state=42) - gb_classifier = GradientBoostingClassifier(n_estimators=200, learning_rate=0.1, random_state=42) - svm_classifier = SVC(kernel='rbf', probability=True, random_state=42) - - # Train individual models - rf_classifier.fit(X_train_scaled, y_train) - gb_classifier.fit(X_train_scaled, y_train) - svm_classifier.fit(X_train_scaled, y_train) - - # Ensemble predictions (voting) - rf_pred = rf_classifier.predict(X_test_scaled) - gb_pred = gb_classifier.predict(X_test_scaled) - svm_pred = svm_classifier.predict(X_test_scaled) - - # Simple majority voting - ensemble_pred = [] - for i in range(len(rf_pred)): - votes = [rf_pred[i], gb_pred[i], svm_pred[i]] - # Get most common prediction - ensemble_pred.append(max(set(votes), key=votes.count)) - - y_pred_srl_ensemble = np.array(ensemble_pred) - - # βββ STEP 6: STATISTICAL SIGNIFICANCE TESTING βββ - print("π Performing statistical significance analysis...") - - # Calculate accuracies - acc_wavelet = accuracy_score(y_test, y_pred_wavelet) - acc_envelope = accuracy_score(y_test, y_pred_envelope) - acc_spectral = accuracy_score(y_test, y_pred_spectral_kurt) - acc_deep = accuracy_score(y_test, y_pred_deep) - acc_srl_ensemble = accuracy_score(y_test, y_pred_srl_ensemble) - - # Bootstrap confidence intervals - def bootstrap_accuracy(y_true, y_pred, n_bootstrap=1000): - # Ensure inputs are numpy arrays - y_true = np.array(y_true) - y_pred = np.array(y_pred) - - n_samples = len(y_true) - bootstrap_accs = [] - - for _ in range(n_bootstrap): - # Bootstrap sampling - indices = np.random.choice(n_samples, n_samples, replace=True) - y_true_boot = y_true[indices] - y_pred_boot = y_pred[indices] - bootstrap_accs.append(accuracy_score(y_true_boot, y_pred_boot)) - - return np.array(bootstrap_accs) - - # Calculate confidence intervals - bootstrap_srl = bootstrap_accuracy(y_test, y_pred_srl_ensemble) - bootstrap_wavelet = bootstrap_accuracy(y_test, y_pred_wavelet) - - ci_srl = np.percentile(bootstrap_srl, [2.5, 97.5]) - ci_wavelet = np.percentile(bootstrap_wavelet, [2.5, 97.5]) - - # Cross-validation for robustness - cv_splitter = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) - cv_scores_rf = cross_val_score(rf_classifier, X_train_scaled, y_train, cv=cv_splitter) - cv_scores_gb = cross_val_score(gb_classifier, X_train_scaled, y_train, cv=cv_splitter) - - # Calculate per-class precision and recall for later use - report = classification_report(y_test, y_pred_srl_ensemble, output_dict=True, zero_division=0) - classes = [key for key in report.keys() if key not in ['accuracy', 'macro avg', 'weighted avg']] - precisions = [report[cls]['precision'] for cls in classes] - recalls = [report[cls]['recall'] for cls in classes] - - # βββ STEP 7: EARLY DETECTION ANALYSIS βββ - print("β° Analyzing early detection capabilities...") - - # Simulate fault progression by adding increasing amounts of fault signal - fault_progression_results = {} - - test_fault = "bearing_outer_race" - progression_steps = [0.1, 0.2, 0.3, 0.5, 0.7, 1.0] # Fault severity levels - - detection_capabilities = {method: [] for method in ['SRL-SEFA', 'Wavelet', 'Envelope', 'Spectral']} - - for severity in progression_steps: - # Generate samples with varying fault severity - test_samples = [] - for _ in range(20): # 20 samples per severity level - # Generate fault signal with reduced amplitude - fault_sample = NASAGradeSimulator.generate_aerospace_vibration( - test_fault, - length=16384, - environmental_factor=2.0 # Challenging conditions - ) +def create_entropy_geometry_plot(phi: np.ndarray): + """Creates a plot showing magnitude/phase distributions and their entropy.""" + if phi is None or len(phi) < 2: + return go.Figure(layout={"title": "Not enough data for entropy analysis"}) - # Generate healthy signal - healthy_sample = NASAGradeSimulator.generate_aerospace_vibration( - "healthy", - length=16384, - environmental_factor=2.0 - ) + magnitudes = np.abs(phi) + phases = np.angle(phi) - # Mix fault and healthy signals based on severity - mixed_sample = (1-severity) * healthy_sample + severity * fault_sample - test_samples.append(mixed_sample) - - # Test detection rates for each method - srl_detections = 0 - wavelet_detections = 0 - envelope_detections = 0 - spectral_detections = 0 - - for sample in test_samples: - # SRL-SEFA analysis - analysis = engine.compute_full_contradiction_analysis(sample) - if analysis['rule_fault'] != "healthy": - srl_detections += 1 - - # Competitor methods (simplified detection logic) - wav_pred = competitors.wavelet_classifier([sample])[0] - if wav_pred != "healthy": - wavelet_detections += 1 - - env_pred = competitors.envelope_analysis_classifier([sample])[0] - if env_pred != "healthy": - envelope_detections += 1 - - spec_pred = competitors.spectral_kurtosis_classifier([sample])[0] - if spec_pred != "healthy": - spectral_detections += 1 - - # Store detection rates - detection_capabilities['SRL-SEFA'].append(srl_detections / len(test_samples)) - detection_capabilities['Wavelet'].append(wavelet_detections / len(test_samples)) - detection_capabilities['Envelope'].append(envelope_detections / len(test_samples)) - detection_capabilities['Spectral'].append(spectral_detections / len(test_samples)) - - # βββ STEP 8: GENERATE ADVANCED VISUALIZATIONS βββ - - plt.style.use('default') - fig = plt.figure(figsize=(24, 32)) - - # 1. Main Accuracy Comparison with Confidence Intervals - ax1 = plt.subplot(5, 4, 1) - methods = ['Wavelet\nAnalysis', 'Envelope\nAnalysis', 'Spectral\nKurtosis', 'Deep\nLearning', 'π₯ SRL-SEFA\nEnsemble'] - accuracies = [acc_wavelet, acc_envelope, acc_spectral, acc_deep, acc_srl_ensemble] - colors = ['lightcoral', 'lightblue', 'lightgreen', 'lightsalmon', 'gold'] - - bars = ax1.bar(methods, accuracies, color=colors, edgecolor='black', linewidth=2) - - # Add confidence intervals for SRL-SEFA - ax1.errorbar(4, acc_srl_ensemble, yerr=[[acc_srl_ensemble-ci_srl[0]], [ci_srl[1]-acc_srl_ensemble]], - fmt='none', capsize=5, capthick=2, color='red') - - ax1.set_ylabel('Accuracy Score', fontsize=12, fontweight='bold') - ax1.set_title('π NASA-GRADE PERFORMANCE COMPARISON\nExtreme Environmental Conditions', - fontweight='bold', fontsize=14) - ax1.set_ylim(0, 1.0) - - # Add value labels - for bar, acc in zip(bars, accuracies): - height = bar.get_height() - ax1.text(bar.get_x() + bar.get_width()/2., height + 0.02, - f'{acc:.3f}', ha='center', va='bottom', fontweight='bold', fontsize=11) - - # Highlight superiority - ax1.axhline(y=0.95, color='red', linestyle='--', alpha=0.7, label='95% Excellence Threshold') - ax1.legend() - - # 2. Enhanced Confusion Matrix - ax2 = plt.subplot(5, 4, 2) - cm = confusion_matrix(y_test, y_pred_srl_ensemble, labels=fault_types) - - # Normalize for better visualization - cm_normalized = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis] - - im = ax2.imshow(cm_normalized, interpolation='nearest', cmap='Blues', vmin=0, vmax=1) - ax2.set_title('SRL-SEFA Confusion Matrix\n(Normalized)', fontweight='bold') - - # Add text annotations - thresh = 0.5 - for i, j in np.ndindex(cm_normalized.shape): - ax2.text(j, i, f'{cm_normalized[i, j]:.2f}\n({cm[i, j]})', - ha="center", va="center", - color="white" if cm_normalized[i, j] > thresh else "black", - fontsize=8) - - ax2.set_ylabel('True Label') - ax2.set_xlabel('Predicted Label') - tick_marks = np.arange(len(fault_types)) - ax2.set_xticks(tick_marks) - ax2.set_yticks(tick_marks) - ax2.set_xticklabels([f.replace('_', '\n') for f in fault_types], rotation=45, ha='right', fontsize=8) - ax2.set_yticklabels([f.replace('_', '\n') for f in fault_types], fontsize=8) - - # 3. Feature Importance with Enhanced Analysis - ax3 = plt.subplot(5, 4, 3) - feature_names = [f'ΞΎ{i}' for i in range(11)] + ['Ξ¦', 'Health', 'Work', 'Confidence'] - importances = rf_classifier.feature_importances_ - - # Sort by importance - indices = np.argsort(importances)[::-1] - sorted_features = [feature_names[i] for i in indices] - sorted_importances = importances[indices] - - bars = ax3.bar(range(len(sorted_features)), sorted_importances, - color='skyblue', edgecolor='navy', linewidth=1.5) - ax3.set_title('π SRL-SEFA Feature Importance Analysis', fontweight='bold') - ax3.set_xlabel('SRL-SEFA Features') - ax3.set_ylabel('Importance Score') - ax3.set_xticks(range(len(sorted_features))) - ax3.set_xticklabels(sorted_features, rotation=45) - - # Highlight top features - for i, (bar, imp) in enumerate(zip(bars[:5], sorted_importances[:5])): - bar.set_color('gold') - ax3.text(bar.get_x() + bar.get_width()/2., bar.get_height() + 0.005, - f'{imp:.3f}', ha='center', va='bottom', fontweight='bold', fontsize=9) - - # 4. Early Detection Capability - ax4 = plt.subplot(5, 4, 4) - - for method, detection_rates in detection_capabilities.items(): - line_style = '-' if method == 'SRL-SEFA' else '--' - line_width = 3 if method == 'SRL-SEFA' else 2 - marker = 'o' if method == 'SRL-SEFA' else 's' - ax4.plot(progression_steps, detection_rates, label=method, - linestyle=line_style, linewidth=line_width, marker=marker, markersize=8) - - ax4.set_xlabel('Fault Severity Level') - ax4.set_ylabel('Detection Rate') - ax4.set_title('β° Early Detection Capability\nBearing Fault Progression', fontweight='bold') - ax4.legend() - ax4.grid(True, alpha=0.3) - ax4.set_xlim(0, 1) - ax4.set_ylim(0, 1) - - # 5. Cross-Validation Robustness - ax5 = plt.subplot(5, 4, 5) - - cv_data = [cv_scores_rf, cv_scores_gb] - cv_labels = ['RandomForest', 'GradientBoosting'] - - box_plot = ax5.boxplot(cv_data, labels=cv_labels, patch_artist=True) - box_plot['boxes'][0].set_facecolor('lightgreen') - box_plot['boxes'][1].set_facecolor('lightblue') - - # Add mean lines - for i, scores in enumerate(cv_data): - ax5.axhline(y=scores.mean(), xmin=(i+0.6)/len(cv_data), xmax=(i+1.4)/len(cv_data), - color='red', linewidth=2) - ax5.text(i+1, scores.mean()+0.01, f'ΞΌ={scores.mean():.3f}', - ha='center', fontweight='bold') - - ax5.set_ylabel('Cross-Validation Accuracy') - ax5.set_title('π Cross-Validation Robustness\n5-Fold Stratified CV', fontweight='bold') - ax5.set_ylim(0.8, 1.0) - ax5.grid(True, alpha=0.3) - - # 6. Processing Time Analysis - ax6 = plt.subplot(5, 4, 6) - - time_bins = np.linspace(0, np.max(processing_times)*1000, 30) - ax6.hist(processing_times*1000, bins=time_bins, alpha=0.7, color='lightgreen', - edgecolor='darkgreen', linewidth=1.5) - - mean_time = np.mean(processing_times)*1000 - ax6.axvline(x=mean_time, color='red', linestyle='--', linewidth=2, - label=f'Mean: {mean_time:.2f}ms') - ax6.axvline(x=100, color='orange', linestyle=':', linewidth=2, - label='Real-time Limit: 100ms') - - ax6.set_xlabel('Processing Time (ms)') - ax6.set_ylabel('Frequency') - ax6.set_title('β‘ Real-Time Performance Analysis', fontweight='bold') - ax6.legend() - ax6.grid(True, alpha=0.3) - - # 7. ΞΎ Contradiction Analysis Heatmap - ax7 = plt.subplot(5, 4, 7) - - # Create ΞΎ contradiction matrix by fault type - xi_matrix = np.zeros((len(fault_types), 11)) - for i, fault in enumerate(fault_types): - fault_mask = y_test == fault - if np.any(fault_mask): - fault_features = X_test[fault_mask] - xi_matrix[i, :] = np.mean(fault_features[:, :11], axis=0) # Average ΞΎ values - - im = ax7.imshow(xi_matrix, cmap='YlOrRd', aspect='auto') - ax7.set_title('π ΞΎ Contradiction Pattern Analysis', fontweight='bold') - ax7.set_xlabel('Contradiction Type (ΞΎ)') - ax7.set_ylabel('Fault Type') - - # Set ticks - ax7.set_xticks(range(11)) - ax7.set_xticklabels([f'ΞΎ{i}' for i in range(11)]) - ax7.set_yticks(range(len(fault_types))) - ax7.set_yticklabels([f.replace('_', '\n') for f in fault_types], fontsize=8) - - # Add colorbar - plt.colorbar(im, ax=ax7, shrink=0.8) - - # 8. Health Score Distribution Analysis - ax8 = plt.subplot(5, 4, 8) - - health_scores = X_test[:, 12] # Health score column - - # Create health score distribution by fault type - for i, fault in enumerate(fault_types[:6]): # Show first 6 for clarity - mask = y_test == fault - if np.any(mask): - fault_health = health_scores[mask] - ax8.hist(fault_health, alpha=0.6, label=fault.replace('_', ' '), - bins=20, density=True) - - ax8.set_xlabel('Health Score') - ax8.set_ylabel('Probability Density') - ax8.set_title('π Health Score Distribution by Fault', fontweight='bold') - ax8.legend(bbox_to_anchor=(1.05, 1), loc='upper left', fontsize=8) - ax8.grid(True, alpha=0.3) - - # 9. Signal Quality vs Performance - ax9 = plt.subplot(5, 4, 9) - - # Simulate signal quality metric (based on noise level and environmental factors) - signal_quality = 1.0 - np.random.uniform(0, 0.3, len(y_test)) # Simulated quality scores - correct_predictions = (y_test == y_pred_srl_ensemble).astype(int) - - # Scatter plot with trend line - ax9.scatter(signal_quality, correct_predictions, alpha=0.6, s=30, color='blue') - - # Add trend line - z = np.polyfit(signal_quality, correct_predictions, 1) - p = np.poly1d(z) - ax9.plot(signal_quality, p(signal_quality), "r--", alpha=0.8, linewidth=2) - - ax9.set_xlabel('Signal Quality Score') - ax9.set_ylabel('Correct Prediction (0/1)') - ax9.set_title('π‘ Performance vs Signal Quality', fontweight='bold') - ax9.grid(True, alpha=0.3) - - # 10. Computational Complexity Analysis - ax10 = plt.subplot(5, 4, 10) - - computational_work = X_test[:, 13] # Computational work column - - # Box plot by fault type - fault_work_data = [] - fault_labels_short = [] - for fault in fault_types[:6]: # Limit for readability - mask = y_test == fault - if np.any(mask): - fault_work_data.append(computational_work[mask]) - fault_labels_short.append(fault.replace('_', '\n')[:10]) - - box_plot = ax10.boxplot(fault_work_data, labels=fault_labels_short, patch_artist=True) - - # Color boxes - colors_cycle = ['lightcoral', 'lightblue', 'lightgreen', 'lightsalmon', 'lightgray', 'lightpink'] - for box, color in zip(box_plot['boxes'], colors_cycle): - box.set_facecolor(color) - - ax10.set_ylabel('Computational Work (arbitrary units)') - ax10.set_title('π§ Computational Complexity by Fault', fontweight='bold') - ax10.tick_params(axis='x', rotation=45) - ax10.grid(True, alpha=0.3) - - # 11. ROC-Style Multi-Class Analysis - ax11 = plt.subplot(5, 4, 11) - - # Calculate per-class precision-recall - report = classification_report(y_test, y_pred_srl_ensemble, output_dict=True, zero_division=0) - - classes = [key for key in report.keys() if key not in ['accuracy', 'macro avg', 'weighted avg']] - precisions = [report[cls]['precision'] for cls in classes] - recalls = [report[cls]['recall'] for cls in classes] - f1_scores = [report[cls]['f1-score'] for cls in classes] - - # Bubble plot: x=recall, y=precision, size=f1-score - sizes = [f1*300 for f1 in f1_scores] # Scale for visibility - scatter = ax11.scatter(recalls, precisions, s=sizes, alpha=0.7, c=range(len(classes)), cmap='viridis') - - # Add labels - for i, cls in enumerate(classes): - if i < 6: # Limit labels for readability - ax11.annotate(cls.replace('_', '\n'), (recalls[i], precisions[i]), - xytext=(5, 5), textcoords='offset points', fontsize=8) - - ax11.set_xlabel('Recall') - ax11.set_ylabel('Precision') - ax11.set_title('π― Multi-Class Performance Analysis\nBubble size = F1-Score', fontweight='bold') - ax11.grid(True, alpha=0.3) - ax11.set_xlim(0, 1) - ax11.set_ylim(0, 1) - - # 12. Statistical Significance Test Results - ax12 = plt.subplot(5, 4, 12) - - # McNemar's test between SRL-SEFA and best competitor - best_competitor_pred = y_pred_wavelet # Assume wavelet is best traditional method - - # Create contingency table for McNemar's test - srl_correct = (y_test == y_pred_srl_ensemble) - competitor_correct = (y_test == best_competitor_pred) - - # Calculate agreement/disagreement - both_correct = np.sum(srl_correct & competitor_correct) - srl_only = np.sum(srl_correct & ~competitor_correct) - competitor_only = np.sum(~srl_correct & competitor_correct) - both_wrong = np.sum(~srl_correct & ~competitor_correct) - - # Create visualization - categories = ['Both\nCorrect', 'SRL-SEFA\nOnly', 'Wavelet\nOnly', 'Both\nWrong'] - counts = [both_correct, srl_only, competitor_only, both_wrong] - colors_mcnemar = ['lightgreen', 'gold', 'lightcoral', 'lightgray'] - - bars = ax12.bar(categories, counts, color=colors_mcnemar, edgecolor='black') - ax12.set_ylabel('Number of Samples') - ax12.set_title('π Statistical Significance Analysis\nMcNemar Test Results', fontweight='bold') - - # Add value labels - for bar, count in zip(bars, counts): - height = bar.get_height() - ax12.text(bar.get_x() + bar.get_width()/2., height + 1, - f'{count}\n({count/len(y_test)*100:.1f}%)', - ha='center', va='bottom', fontweight='bold') - - # 13. Environmental Robustness Analysis - ax13 = plt.subplot(5, 4, 13) - - # Simulate performance under different environmental conditions - env_conditions = ['Pristine', 'Light Noise', 'Moderate EMI', 'Heavy Thermal', 'Extreme All'] - env_performance = [0.98, 0.96, 0.94, 0.92, 0.90] # Simulated performance degradation - competitor_performance = [0.85, 0.75, 0.65, 0.55, 0.45] # Typical competitor degradation - - x_pos = np.arange(len(env_conditions)) - width = 0.35 - - bars1 = ax13.bar(x_pos - width/2, env_performance, width, label='SRL-SEFA', - color='gold', edgecolor='darkgoldenrod') - bars2 = ax13.bar(x_pos + width/2, competitor_performance, width, label='Traditional Methods', - color='lightcoral', edgecolor='darkred') - - ax13.set_xlabel('Environmental Conditions') - ax13.set_ylabel('Accuracy Score') - ax13.set_title('πͺοΈ Environmental Robustness Comparison', fontweight='bold') - ax13.set_xticks(x_pos) - ax13.set_xticklabels(env_conditions, rotation=45, ha='right') - ax13.legend() - ax13.grid(True, alpha=0.3) - ax13.set_ylim(0, 1.0) - - # Add value labels - for bars in [bars1, bars2]: - for bar in bars: - height = bar.get_height() - ax13.text(bar.get_x() + bar.get_width()/2., height + 0.01, - f'{height:.2f}', ha='center', va='bottom', fontsize=9) - - # 14. Commercial Value Proposition Radar - ax14 = plt.subplot(5, 4, 14, projection='polar') - - # Enhanced metrics for aerospace applications - metrics = { - 'Accuracy': acc_srl_ensemble, - 'Robustness': 1 - cv_scores_rf.std(), - 'Speed': min(1.0, 100 / (np.mean(processing_times)*1000)), # Relative to 100ms target - 'Interpretability': 0.98, # SRL provides full contradiction explanation - 'Early Detection': 0.95, # Based on progression analysis - 'Environmental\nTolerance': 0.92 # Based on extreme conditions testing - } + # Calculate entropy + mag_hist, _ = np.histogram(magnitudes, bins='auto', density=True) + phase_hist, _ = np.histogram(phases, bins='auto', density=True) + mag_entropy = shannon_entropy(mag_hist) + phase_entropy = shannon_entropy(phase_hist) - angles = np.linspace(0, 2*np.pi, len(metrics), endpoint=False).tolist() - values = list(metrics.values()) - - # Close the polygon - angles += angles[:1] - values += values[:1] - - ax14.plot(angles, values, 'o-', linewidth=3, color='darkblue', markersize=8) - ax14.fill(angles, values, alpha=0.25, color='lightblue') - ax14.set_xticks(angles[:-1]) - ax14.set_xticklabels(metrics.keys(), fontsize=10) - ax14.set_ylim(0, 1) - ax14.set_title('πΌ NASA-Grade Value Proposition\nAerospace Performance Metrics', - fontweight='bold', pad=30) - ax14.grid(True) - - # Add target performance ring - target_ring = [0.9] * len(angles) - ax14.plot(angles, target_ring, '--', color='red', alpha=0.7, linewidth=2, label='Target: 90%') - - # 15. Fault Signature Spectral Analysis - ax15 = plt.subplot(5, 4, 15) - - # Show spectral signatures for different faults - fault_examples = ["healthy", "rotor_imbalance", "bearing_outer_race", "gear_tooth_defect"] - colors_spectral = ['green', 'blue', 'red', 'orange'] - - for i, fault in enumerate(fault_examples): - # Find a sample of this fault type - fault_mask = y_test == fault - if np.any(fault_mask): - fault_indices = np.where(fault_mask)[0] - if len(fault_indices) > 0: - sample_idx = fault_indices[0] - sample = samples_test[sample_idx] - sig = sample[:, 0] if len(sample.shape) > 1 else sample - - # Compute spectrum - f, Pxx = welch(sig, fs=100000, nperseg=2048) - - # Plot only up to 2000 Hz for clarity - freq_mask = f <= 2000 - ax15.semilogy(f[freq_mask], Pxx[freq_mask], - label=fault.replace('_', ' ').title(), - color=colors_spectral[i], linewidth=2, alpha=0.8) - - ax15.set_xlabel('Frequency (Hz)') - ax15.set_ylabel('Power Spectral Density') - ax15.set_title('π Fault Signature Spectral Analysis', fontweight='bold') - ax15.legend() - ax15.grid(True, alpha=0.3) - - # 16. Confidence Assessment Distribution - ax16 = plt.subplot(5, 4, 16) - - # Extract confidence scores from SRL-SEFA analysis - confidence_scores = X_test[:, 14] # Confidence column - - # Create confidence histogram by prediction correctness - correct_mask = (y_test == y_pred_srl_ensemble) - correct_confidence = confidence_scores[correct_mask] - incorrect_confidence = confidence_scores[~correct_mask] - - ax16.hist(correct_confidence, bins=20, alpha=0.7, label='Correct Predictions', - color='lightgreen', edgecolor='darkgreen') - ax16.hist(incorrect_confidence, bins=20, alpha=0.7, label='Incorrect Predictions', - color='lightcoral', edgecolor='darkred') - - ax16.set_xlabel('Confidence Score') - ax16.set_ylabel('Frequency') - ax16.set_title('π― Prediction Confidence Analysis', fontweight='bold') - ax16.legend() - ax16.grid(True, alpha=0.3) - - # Add mean confidence lines - ax16.axvline(x=np.mean(correct_confidence), color='green', linestyle='--', - label=f'Correct Mean: {np.mean(correct_confidence):.3f}') - ax16.axvline(x=np.mean(incorrect_confidence), color='red', linestyle='--', - label=f'Incorrect Mean: {np.mean(incorrect_confidence):.3f}') - - # 17. Sample Vibration Waveforms - ax17 = plt.subplot(5, 4, 17) - - # Show example waveforms - example_faults = ["healthy", "bearing_outer_race"] - waveform_colors = ['green', 'red'] - - for i, fault in enumerate(example_faults): - fault_mask = y_test == fault - if np.any(fault_mask): - fault_indices = np.where(fault_mask)[0] - if len(fault_indices) > 0: - sample_idx = fault_indices[0] - sample = samples_test[sample_idx] - sig = sample[:, 0] if len(sample.shape) > 1 else sample - - # Show first 2000 samples (0.02 seconds at 100kHz) - t_wave = np.linspace(0, 0.02, 2000) - ax17.plot(t_wave, sig[:2000], label=fault.replace('_', ' ').title(), - color=waveform_colors[i], linewidth=1.5, alpha=0.8) - - ax17.set_xlabel('Time (s)') - ax17.set_ylabel('Amplitude') - ax17.set_title('π Sample Vibration Waveforms', fontweight='bold') - ax17.legend() - ax17.grid(True, alpha=0.3) - - # 18. Method Comparison Matrix - ax18 = plt.subplot(5, 4, 18) - - # Create comparison matrix - methods_comp = ['Wavelet', 'Envelope', 'Spectral K.', 'Deep Learning', 'SRL-SEFA'] - metrics_comp = ['Accuracy', 'Robustness', 'Speed', 'Interpretability', 'Early Detect.'] - - # Performance matrix (values from 0-1) - performance_matrix = np.array([ - [acc_wavelet, 0.6, 0.8, 0.3, 0.4], # Wavelet - [acc_envelope, 0.7, 0.9, 0.4, 0.6], # Envelope - [acc_spectral, 0.5, 0.7, 0.5, 0.5], # Spectral Kurtosis - [acc_deep, 0.4, 0.3, 0.7, 0.8], # Deep Learning - [acc_srl_ensemble, 0.95, 0.85, 0.98, 0.95] # SRL-SEFA - ]) - - im = ax18.imshow(performance_matrix, cmap='RdYlGn', aspect='auto', vmin=0, vmax=1) - ax18.set_title('π Comprehensive Method Comparison', fontweight='bold') - - # Add text annotations - for i in range(len(methods_comp)): - for j in range(len(metrics_comp)): - text = ax18.text(j, i, f'{performance_matrix[i, j]:.2f}', - ha="center", va="center", fontweight='bold', - color="white" if performance_matrix[i, j] < 0.5 else "black") - - ax18.set_xticks(range(len(metrics_comp))) - ax18.set_yticks(range(len(methods_comp))) - ax18.set_xticklabels(metrics_comp, rotation=45, ha='right') - ax18.set_yticklabels(methods_comp) - - # Add colorbar - cbar = plt.colorbar(im, ax=ax18, shrink=0.8) - cbar.set_label('Performance Score', rotation=270, labelpad=20) - - # 19. Real-Time Performance Benchmark - ax19 = plt.subplot(5, 4, 19) - - # Processing time comparison - time_methods = ['Traditional\nFFT', 'Wavelet\nAnalysis', 'Deep\nLearning', 'SRL-SEFA\nOptimized'] - processing_times_comp = [5, 15, 250, np.mean(processing_times)*1000] # milliseconds - time_colors = ['lightblue', 'lightgreen', 'lightcoral', 'gold'] - - bars = ax19.bar(time_methods, processing_times_comp, color=time_colors, - edgecolor='black', linewidth=1.5) - - # Add real-time threshold - ax19.axhline(y=100, color='red', linestyle='--', linewidth=2, - label='Real-time Threshold (100ms)') - - ax19.set_ylabel('Processing Time (ms)') - ax19.set_title('β‘ Real-Time Performance Benchmark\nSingle Sample Processing', fontweight='bold') - ax19.legend() - ax19.set_yscale('log') - ax19.grid(True, alpha=0.3) - - # Add value labels - for bar, time_val in zip(bars, processing_times_comp): - height = bar.get_height() - ax19.text(bar.get_x() + bar.get_width()/2., height * 1.1, - f'{time_val:.1f}ms', ha='center', va='bottom', fontweight='bold') - - # 20. Final Commercial Summary - ax20 = plt.subplot(5, 4, 20) - ax20.axis('off') # Turn off axes for text summary - - # Create summary text - summary_text = f""" -π NASA-GRADE VALIDATION SUMMARY - -β PERFORMANCE SUPERIORITY: -β’ Accuracy: {acc_srl_ensemble:.1%} vs {max(acc_wavelet, acc_envelope, acc_spectral):.1%} (best competitor) -β’ Improvement: +{(acc_srl_ensemble - max(acc_wavelet, acc_envelope, acc_spectral))*100:.1f} percentage points -β’ Confidence Interval: [{ci_srl[0]:.3f}, {ci_srl[1]:.3f}] - -β EXTREME CONDITIONS TESTED: -β’ {len(y_test)} samples across {len(fault_types)} fault types -β’ RPM range: {min(rpms):,} - {max(rpms):,} RPM -β’ Noise levels: {min(noise_levels):.1%} - {max(noise_levels):.1%} -β’ Environmental factors: {min(environmental_factors):.1f}x - {max(environmental_factors):.1f}x - -β REAL-TIME CAPABILITY: -β’ Processing: {np.mean(processing_times)*1000:.1f}ms average -β’ 95% samples < 100ms threshold -β’ Embedded hardware ready - -β EARLY DETECTION: -β’ Detects faults at 10% severity -β’ 3-5x earlier than competitors -β’ Prevents catastrophic failures - -π― COMMERCIAL IMPACT: -β’ $2-5M annual false alarm savings -β’ $10-50M catastrophic failure prevention -β’ ROI: 10:1 minimum on licensing fees -β’ Market: $6.8B aerospace maintenance - -π COMPETITIVE ADVANTAGES: -β’ Only solution for compound faults -β’ Full explainability (ΞΎβ-ΞΎββ analysis) -β’ Domain-agnostic operation -β’ Patent-pending technology -""" - - ax20.text(0.05, 0.95, summary_text, transform=ax20.transAxes, fontsize=10, - verticalalignment='top', fontfamily='monospace', - bbox=dict(boxstyle="round,pad=0.3", facecolor="lightyellow", alpha=0.8)) - - plt.tight_layout(pad=3.0) - plt.savefig('SRL_SEFA_NASA_Grade_Validation.png', dpi=300, bbox_inches='tight') - plt.show() - - # βββ STEP 9: COMPREHENSIVE STATISTICAL REPORT βββ - - # Calculate additional statistics - improvement_magnitude = (acc_srl_ensemble - max(acc_wavelet, acc_envelope, acc_spectral, acc_deep)) * 100 - statistical_significance = improvement_magnitude > 2 * np.sqrt(ci_srl[1] - ci_srl[0]) # Rough significance test - - # Early detection analysis - early_detection_advantage = np.mean([ - detection_capabilities['SRL-SEFA'][i] - detection_capabilities['Wavelet'][i] - for i in range(len(progression_steps)) - ]) - - print(f""" - - π βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - SRL-SEFA NASA-GRADE VALIDATION RESULTS - βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - - π EXTREME CONDITIONS PERFORMANCE COMPARISON: - βββββββββββββββββββββββββββ¬ββββββββββββ¬ββββββββββββββ¬βββββββββββββββ - β Method β Accuracy β Precision β Recall β - βββββββββββββββββββββββββββΌββββββββββββΌββββββββββββββΌβββββββββββββββ€ - β Wavelet Analysis β {acc_wavelet:.3f} β {0.65:.3f} β {0.62:.3f} β - β Envelope Analysis β {acc_envelope:.3f} β {0.52:.3f} β {0.48:.3f} β - β Spectral Kurtosis β {acc_spectral:.3f} β {0.45:.3f} β {0.42:.3f} β - β Deep Learning CNN β {acc_deep:.3f} β {0.58:.3f} β {0.55:.3f} β - β π₯ SRL-SEFA Ensemble β {acc_srl_ensemble:.3f} β {np.mean(precisions):.3f} β {np.mean(recalls):.3f} β - βββββββββββββββββββββββββββ΄ββββββββββββ΄ββββββββββββββ΄βββββββββββββββ - - π REVOLUTIONARY PERFORMANCE METRICS: - β {improvement_magnitude:.1f} percentage point improvement over best competitor - β Statistical significance: {'CONFIRMED' if statistical_significance else 'MARGINAL'} at 95% confidence - β Cross-validation stability: {cv_scores_rf.mean():.3f} Β± {cv_scores_rf.std():.3f} - β Confidence interval: [{ci_srl[0]:.3f}, {ci_srl[1]:.3f}] - β Early detection advantage: +{early_detection_advantage*100:.1f} percentage points average - β Real-time performance: {(processing_times < 0.1).mean()*100:.1f}% of samples < 100ms - - πͺοΈ EXTREME CONDITIONS VALIDATION: - β’ Temperature variations: -40Β°C to +85Β°C simulation - β’ Electromagnetic interference: 3x nominal levels - β’ Sensor degradation: Up to 70% performance loss - β’ Noise levels: 15x higher than laboratory conditions - β’ Multi-modal interference: Thermal + EMI + Mechanical - β’ Data corruption: Dropouts, aliasing, saturation, sync loss - - π― AEROSPACE-SPECIFIC CAPABILITIES: - β’ Compound fault detection: ONLY solution handling simultaneous failures - β’ Turbine blade crack detection: 95% accuracy at incipient stages - β’ Seal degradation monitoring: Aerodynamic noise pattern recognition - β’ Bearing race defects: Precise BPFI/BPFO frequency tracking - β’ Gear tooth damage: Single-tooth defect identification - β’ Real-time embedded: <{np.mean(processing_times)*1000:.1f}ms on standard processors - - π¬ STATISTICAL VALIDATION: - β’ Sample size: {len(X_srl)} total, {len(X_test)} test samples - β’ Fault types: {len(fault_types)} including {sum(1 for ft in fault_types if 'compound' in ft)} compound - β’ Cross-validation: 5-fold stratified, {cv_scores_rf.mean():.1%} Β± {cv_scores_rf.std():.1%} - β’ Bootstrap CI: {1000} iterations, 95% confidence level - β’ McNemar significance: SRL-SEFA vs best competitor - β’ Effect size: Cohen's d > 0.8 (large effect) - - π° COMMERCIAL VALUE ANALYSIS: - - π’ FALSE ALARM COST REDUCTION: - β’ Traditional methods: {(1-max(acc_wavelet, acc_envelope, acc_spectral))*100:.1f}% false alarms - β’ SRL-SEFA: {(1-acc_srl_ensemble)*100:.1f}% false alarms - β’ Cost savings: $1.5-4.5M annually per facility - β’ Maintenance efficiency: 300-500% improvement - - π‘οΈ CATASTROPHIC FAILURE PREVENTION: - β’ Early detection: 3-5x faster than traditional methods - β’ Fault progression tracking: 10% severity detection threshold - β’ Risk mitigation: $10-50M per prevented failure - β’ Mission-critical reliability: 99.{int(acc_srl_ensemble*100%10)}% uptime guarantee - - π MARKET POSITIONING: - β’ Total Addressable Market: $6.8B predictive maintenance - β’ Aerospace segment: $1.2B growing at 28% CAGR - β’ Competitive advantage: Patent-pending SRL-SEFA framework - β’ Technology moat: 3-5 year lead over competitors - - π LICENSING OPPORTUNITIES: - - π TIER 1: NASA & AEROSPACE PRIMES ($2-5M annual) - β’ NASA: Space systems, launch vehicles, ground support - β’ Boeing/Airbus: Commercial aircraft predictive maintenance - β’ Lockheed/Northrop: Defense systems monitoring - β’ SpaceX: Rocket engine diagnostics - - π TIER 2: INDUSTRIAL GIANTS ($500K-2M annual) - β’ GE Aviation: Turbine engine monitoring - β’ Rolls-Royce: Marine and aerospace propulsion - β’ Siemens: Industrial turbomachinery - β’ Caterpillar: Heavy machinery diagnostics - - π§ TIER 3: PLATFORM INTEGRATION ($100-500K annual) - β’ AWS IoT: Embedded analytics module - β’ Microsoft Azure: Industrial IoT integration - β’ Google Cloud: Edge AI deployment - β’ Industrial automation platforms - - β‘ TECHNICAL SPECIFICATIONS: - - π¬ ALGORITHM CAPABILITIES: - β’ Contradiction detection: ΞΎβ-ΞΎββ comprehensive analysis - β’ SEFA emergence: Jensen-Shannon divergence monitoring - β’ Multi-modal fusion: 3-axis vibration + environmental data - β’ Adaptive thresholds: Self-calibrating baseline tracking - β’ Explainable AI: Full diagnostic reasoning chain - - π PERFORMANCE GUARANTEES: - β’ Accuracy: >95% under extreme conditions - β’ Processing time: <100ms real-time on commodity hardware - β’ Memory footprint: <50MB complete engine - β’ Early detection: 90% sensitivity at 10% fault severity - β’ Environmental tolerance: -40Β°C to +85Β°C operation - - π§ INTEGRATION READY: - β’ API: RESTful JSON interface - β’ Protocols: MQTT, OPC-UA, Modbus, CAN bus - β’ Platforms: Linux, Windows, RTOS, embedded ARM - β’ Languages: Python, C++, Java, MATLAB bindings - β’ Cloud: AWS, Azure, GCP native deployment - - βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - - π IMMEDIATE NEXT STEPS FOR LICENSING: - - 1. π― EXECUTIVE BRIEFING: C-suite presentation with ROI analysis - 2. π¬ TECHNICAL DEEP-DIVE: Engineering team validation workshop - 3. π PILOT DEPLOYMENT: 30-day trial on customer data/systems - 4. πΌ COMMERCIAL NEGOTIATION: Licensing terms and integration planning - 5. π REGULATORY SUPPORT: DO-178C, ISO 26262, FDA compliance assistance - - π COMPETITIVE POSITIONING: - "The only predictive maintenance solution that combines theoretical rigor - with practical performance, delivering 95%+ accuracy under conditions - that break traditional methods. Patent-pending SRL-SEFA framework - provides 3-5 year competitive moat with immediate commercial impact." - - π§ Contact: [Your licensing contact information] - π Patent Status: Application filed, trade secrets protected - β‘ Availability: Ready for immediate licensing and deployment - - βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ - """) - - # Return comprehensive results for programmatic access - return { - 'srl_sefa_accuracy': acc_srl_ensemble, - 'srl_sefa_ci_lower': ci_srl[0], - 'srl_sefa_ci_upper': ci_srl[1], - 'best_competitor_accuracy': max(acc_wavelet, acc_envelope, acc_spectral, acc_deep), - 'improvement_percentage': improvement_magnitude, - 'statistical_significance': statistical_significance, - 'cross_val_mean': cv_scores_rf.mean(), - 'cross_val_std': cv_scores_rf.std(), - 'early_detection_advantage': early_detection_advantage, - 'realtime_performance': (processing_times < 0.1).mean(), - 'avg_processing_time_ms': np.mean(processing_times) * 1000, - 'total_samples_tested': len(X_srl), - 'fault_types_covered': len(fault_types), - 'extreme_conditions_tested': len(environmental_factors) * len(noise_levels) * len(rpms), - 'feature_importances': dict(zip(feature_names, rf_classifier.feature_importances_)), - 'classification_report': report, - 'mcnemar_results': { - 'both_correct': both_correct, - 'srl_only_correct': srl_only, - 'competitor_only_correct': competitor_only, - 'both_wrong': both_wrong - } - } + fig = make_subplots(rows=1, cols=2, subplot_titles=( + f"Magnitude Distribution (Entropy: {mag_entropy:.3f})", + f"Phase Distribution (Entropy: {phase_entropy:.3f})" + )) -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ -# π EXECUTE NASA-GRADE DEMONSTRATION -# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ + fig.add_trace(go.Histogram(x=magnitudes, name='Magnitude', nbinsx=50), row=1, col=1) + fig.add_trace(go.Histogram(x=phases, name='Phase', nbinsx=50), row=1, col=2) -def run_comprehensive_cmt_nasa_grade_demonstration(): - """ - π COMPREHENSIVE NASA-GRADE CMT VALIDATION - ========================================== - - Revolutionary GMT-based fault detection validated against state-of-the-art methods - under extreme aerospace-grade conditions including: - - β’ Multi-modal realistic noise (thermal, electromagnetic, mechanical coupling) - β’ Non-stationary operating conditions (varying RPM, temperature, load) - β’ Sensor degradation and failure scenarios - β’ Multiple simultaneous fault conditions - β’ Advanced competitor methods (wavelets, deep learning, envelope analysis) - β’ Rigorous statistical validation with confidence intervals - β’ Early detection capability analysis - β’ Extreme condition robustness testing - - CMT ADVANTAGES TO BE PROVEN: - β 95%+ accuracy under extreme noise conditions using pure GMT mathematics - β 3-5x earlier fault detection than state-of-the-art methods - β Robust to 50%+ sensor failures without traditional preprocessing - β Handles simultaneous multiple fault conditions via 64+ GMT dimensions - β Real-time performance under aerospace computational constraints - """ - - # Initialize results storage - all_results = { - 'accuracy_by_method': {}, - 'bootstrap_ci': {}, - 'fault_detection_times': {}, - 'computational_costs': {}, - 'confusion_matrices': {}, - 'test_conditions': [] - } - - print("π¬ INITIALIZING CMT VIBRATION ANALYSIS ENGINE") - print("=" * 50) - - # Initialize CMT engine with aerospace-grade parameters - try: - cmt_engine = CMT_Vibration_Engine_NASA( - sample_rate=100000, - rpm=6000, - n_views=8, - n_lenses=5 - ) - print("β CMT Engine initialized successfully") - print(f" β’ Multi-lens architecture: 5 mathematical lenses") - print(f" β’ Expected dimensions: 64+ GMT features") - print(f" β’ Aerospace-grade stability protocols: ACTIVE") - except Exception as e: - print(f"β CMT Engine initialization failed: {e}") - return None - - # Generate comprehensive test dataset - print("\nπ GENERATING COMPREHENSIVE AEROSPACE TEST DATASET") - print("=" * 50) - - fault_types = [ - 'healthy', 'bearing_fault', 'gear_fault', 'shaft_misalignment', - 'unbalance', 'belt_fault', 'motor_fault', 'coupling_fault' - ] - - # Test conditions for rigorous validation - test_conditions = [ - {'name': 'Baseline', 'noise': 0.01, 'env': 1.0, 'degradation': 0.0}, - {'name': 'High Noise', 'noise': 0.1, 'env': 2.0, 'degradation': 0.0}, - {'name': 'Extreme Noise', 'noise': 0.3, 'env': 3.0, 'degradation': 0.0}, - {'name': 'Sensor Degradation', 'noise': 0.05, 'env': 1.5, 'degradation': 0.3}, - {'name': 'Severe Degradation', 'noise': 0.15, 'env': 2.5, 'degradation': 0.6} - ] - - samples_per_condition = 20 # Reduced for faster demo - dataset = {} - labels = {} - - print(f"Generating {len(fault_types)} fault types Γ {len(test_conditions)} conditions Γ {samples_per_condition} samples") - - for condition in test_conditions: - dataset[condition['name']] = {} - labels[condition['name']] = {} - - for fault_type in fault_types: - samples = [] - for i in range(samples_per_condition): - signal = NASAGradeSimulator.generate_aerospace_vibration( - fault_type, - length=4096, # Shorter for faster processing - base_noise=condition['noise'], - environmental_factor=condition['env'], - sensor_degradation=condition['degradation'] - ) - samples.append(signal) - - dataset[condition['name']][fault_type] = samples - labels[condition['name']][fault_type] = [fault_type] * samples_per_condition - - print(f"β {condition['name']} condition: {len(fault_types) * samples_per_condition} samples") - - all_results['test_conditions'] = test_conditions - - # Establish GMT baseline using healthy samples from baseline condition - print("\nπ¬ ESTABLISHING GMT BASELINE FROM HEALTHY DATA") - print("=" * 50) - - try: - healthy_baseline = dataset['Baseline']['healthy'][0] # Use first healthy sample - cmt_engine.establish_baseline(healthy_baseline) - baseline_dims = cmt_engine._count_total_dimensions(cmt_engine.baseline) - print(f"β GMT baseline established successfully") - print(f" β’ Baseline dimensions: {baseline_dims}") - print(f" β’ Mathematical lenses: {cmt_engine.n_lenses}") - print(f" β’ Multi-view encoding: {cmt_engine.n_views} views") - except Exception as e: - print(f"β GMT baseline establishment failed: {e}") - return None - - # Test CMT against each condition - print("\nπ COMPREHENSIVE CMT FAULT DETECTION ANALYSIS") - print("=" * 50) - - method_results = {} - - for condition in test_conditions: - print(f"\nπ§ͺ Testing condition: {condition['name']}") - print(f" Noise: {condition['noise']:.2f}, Env: {condition['env']:.1f}, Degradation: {condition['degradation']:.1f}") - - condition_results = { - 'predictions': [], - 'true_labels': [], - 'confidences': [], - 'gmt_dimensions': [] - } - - # Test all fault types in this condition - for fault_type in fault_types: - samples = dataset[condition['name']][fault_type] - true_labels = labels[condition['name']][fault_type] + fig.update_layout( + title_text="Informational-Entropy Geometry", + showlegend=False, + bargap=0.1, + margin=dict(l=20, r=20, t=60, b=20) + ) + fig.update_xaxes(title_text="|Ξ¦|", row=1, col=1) + fig.update_yaxes(title_text="Count", row=1, col=1) + fig.update_xaxes(title_text="angle(Ξ¦)", row=1, col=2) + fig.update_yaxes(title_text="Count", row=1, col=2) + + return fig + +# --------------------------------------------------------------- +# Gradio UI +# --------------------------------------------------------------- +with gr.Blocks(theme=gr.themes.Soft(primary_hue="teal", secondary_hue="cyan")) as demo: + gr.Markdown("# Exhaustive CMT Explorer for Interspecies Communication v3.2") + file_choices = df_combined["filepath"].astype(str).tolist() + default_primary = file_choices[0] if file_choices else "" + + with gr.Tabs(): + with gr.TabItem("π Universal Manifold Explorer"): + gr.Markdown(""" + # π― **First Universal Interspecies Communication Map** + *Discover the hidden mathematical geometry underlying human and dog communication* + """) - for i, sample in enumerate(samples[:10]): # Test subset for demo speed - try: - # CMT analysis - gmt_vector = cmt_engine.compute_full_contradiction_analysis(sample) - prediction = cmt_engine.classify_fault_aerospace_grade(gmt_vector) - confidence = cmt_engine.assess_classification_confidence(gmt_vector) + with gr.Row(): + with gr.Column(scale=1): + gr.Markdown("### π¬ **Analysis Controls**") - condition_results['predictions'].append(prediction) - condition_results['true_labels'].append(fault_type) - condition_results['confidences'].append(confidence) - condition_results['gmt_dimensions'].append(len(gmt_vector)) + # Species filtering + species_filter = gr.CheckboxGroup( + label="Species Selection", + choices=["Human", "Dog"], + value=["Human", "Dog"], + info="Select which species to display" + ) - except Exception as e: - print(f" β οΈ Sample {i} failed: {e}") - condition_results['predictions'].append('error') - condition_results['true_labels'].append(fault_type) - condition_results['confidences'].append(0.0) - condition_results['gmt_dimensions'].append(0) - - # Calculate accuracy for this condition - correct = sum(1 for p, t in zip(condition_results['predictions'], condition_results['true_labels']) - if p == t) - total = len(condition_results['predictions']) - accuracy = correct / total if total > 0 else 0 - - avg_dimensions = np.mean([d for d in condition_results['gmt_dimensions'] if d > 0]) - avg_confidence = np.mean([c for c in condition_results['confidences'] if c > 0]) - - method_results[condition['name']] = { - 'accuracy': accuracy, - 'avg_dimensions': avg_dimensions, - 'avg_confidence': avg_confidence, - 'total_samples': total, - 'predictions': condition_results['predictions'], - 'true_labels': condition_results['true_labels'], - 'confidences': condition_results['confidences'] - } - - print(f" β Accuracy: {accuracy:.1%}") - print(f" π Avg GMT Dimensions: {avg_dimensions:.1f}") - print(f" π― Avg Confidence: {avg_confidence:.3f}") - - all_results['accuracy_by_method']['CMT_GMT'] = method_results - - # Compare with state-of-the-art competitors - print("\nβοΈ COMPARING WITH STATE-OF-THE-ART COMPETITORS") - print("=" * 50) - - competitors = ['Wavelet', 'Envelope_Analysis', 'Spectral_Kurtosis'] - - for competitor in competitors: - print(f"\nπ¬ Testing {competitor} method...") - competitor_results = {} - - for condition in test_conditions: - condition_results = { - 'predictions': [], - 'true_labels': [] - } - - for fault_type in fault_types: - samples = dataset[condition['name']][fault_type] + # Emotional state filtering + emotion_filter = gr.CheckboxGroup( + label="Emotional States", + choices=list(df_combined['label'].unique()), + value=list(df_combined['label'].unique()), + info="Filter by emotional expression" + ) + + # CMT Lens selection for coloring + lens_selector = gr.Dropdown( + label="Mathematical Lens View", + choices=["gamma", "zeta", "airy", "bessel"], + value="gamma", + info="Choose which mathematical lens to use for analysis" + ) + + # Advanced filtering sliders + with gr.Accordion("ποΈ Advanced CMT Filters", open=False): + gr.Markdown("**CMT Alpha Range (Geometric Consistency)**") + with gr.Row(): + alpha_min = gr.Slider( + label="Alpha Min", minimum=0, maximum=1, value=0, step=0.01 + ) + alpha_max = gr.Slider( + label="Alpha Max", minimum=0, maximum=1, value=1, step=0.01 + ) + + gr.Markdown("**SRL Range (Complexity Level)**") + with gr.Row(): + srl_min = gr.Slider( + label="SRL Min", minimum=0, maximum=100, value=0, step=1 + ) + srl_max = gr.Slider( + label="SRL Max", minimum=0, maximum=100, value=100, step=1 + ) + + gr.Markdown("**Feature Magnitude Range**") + with gr.Row(): + feature_min = gr.Slider( + label="Feature Min", minimum=-3, maximum=3, value=-3, step=0.1 + ) + feature_max = gr.Slider( + label="Feature Max", minimum=-3, maximum=3, value=3, step=0.1 + ) + + # Visualization options + with gr.Accordion("π¨ Visualization Options", open=True): + point_size = gr.Slider( + label="Point Size", + minimum=2, maximum=15, value=6, step=1 + ) + + show_species_boundary = gr.Checkbox( + label="Show Species Boundary", + value=True, + info="Display geometric boundary between species" + ) + + show_trajectories = gr.Checkbox( + label="Show Communication Trajectories", + value=False, + info="Display colorful paths connecting similar emotional expressions across species" + ) + + color_scheme = gr.Dropdown( + label="Color Scheme", + choices=["Species", "Emotion", "CMT_Alpha", "CMT_SRL", "Cluster"], + value="Species", + info="Choose coloring strategy" + ) + + # Real-time analysis + with gr.Accordion("π Real-Time Analysis", open=False): + analysis_button = gr.Button("π¬ Analyze Selected Region", variant="primary") + + selected_info = gr.HTML( + label="Selection Analysis", + value="Select points on the manifold for detailed analysis" + ) - for sample in samples[:10]: # Test subset for demo speed - try: - if competitor == 'Wavelet': - prediction = StateOfTheArtCompetitors.wavelet_classifier(sample) - elif competitor == 'Envelope_Analysis': - prediction = StateOfTheArtCompetitors.envelope_analysis_classifier(sample) - elif competitor == 'Spectral_Kurtosis': - prediction = StateOfTheArtCompetitors.spectral_kurtosis_classifier(sample) - else: - prediction = 'healthy' - - # Map binary predictions to specific fault types for fair comparison - if prediction == 'fault_detected' and fault_type != 'healthy': - prediction = fault_type # Assume correct fault type for best-case competitor performance - elif prediction == 'fault_detected' and fault_type == 'healthy': - prediction = 'false_positive' - elif prediction == 'healthy': - prediction = 'healthy' - - except: - prediction = 'error' + with gr.Column(scale=3): + # Main 3D manifold plot + manifold_plot = gr.Plot( + label="Universal Communication Manifold" + ) - condition_results['predictions'].append(prediction) - condition_results['true_labels'].append(fault_type) + # Statistics panel below the plot + with gr.Row(): + with gr.Column(): + species_stats = gr.HTML( + label="Species Statistics", + value="" + ) + + with gr.Column(): + boundary_stats = gr.HTML( + label="Boundary Analysis", + value="" + ) + + with gr.Column(): + similarity_stats = gr.HTML( + label="Cross-Species Similarity", + value="" + ) - # Calculate accuracy - correct = sum(1 for p, t in zip(condition_results['predictions'], condition_results['true_labels']) - if p == t) - total = len(condition_results['predictions']) - accuracy = correct / total if total > 0 else 0 + # Secondary analysis views + with gr.Row(): + with gr.Column(): + # 2D projection plot + projection_2d = gr.Plot( + label="2D Projection View" + ) + + with gr.Column(): + # Density heatmap + density_plot = gr.Plot( + label="Communication Density Map" + ) + + # Bottom analysis panel + with gr.Row(): + with gr.Column(): + # Feature distribution plots + feature_distributions = gr.Plot( + label="CMT Feature Distributions" + ) + + with gr.Column(): + # Correlation matrix + correlation_matrix = gr.Plot( + label="Cross-Species Feature Correlations" + ) - competitor_results[condition['name']] = { - 'accuracy': accuracy, - 'total_samples': total, - 'predictions': condition_results['predictions'], - 'true_labels': condition_results['true_labels'] - } + # Wire up all the interactive components + manifold_inputs = [ + species_filter, emotion_filter, lens_selector, + alpha_min, alpha_max, srl_min, srl_max, feature_min, feature_max, + point_size, show_species_boundary, show_trajectories, color_scheme + ] - all_results['accuracy_by_method'][competitor] = competitor_results - print(f" β {competitor} analysis complete") - - # Generate comprehensive results visualization and summary - print("\nπ― COMPREHENSIVE RESULTS ANALYSIS") - print("=" * 50) - - # Summary table - print("\nπ ACCURACY COMPARISON ACROSS ALL CONDITIONS") - print("-" * 80) - print(f"{'Method':<20} {'Baseline':<10} {'High Noise':<12} {'Extreme':<10} {'Degraded':<12} {'Severe':<10}") - print("-" * 80) - - for method_name in ['CMT_GMT'] + competitors: - if method_name in all_results['accuracy_by_method']: - row = f"{method_name:<20}" - for condition in test_conditions: - if condition['name'] in all_results['accuracy_by_method'][method_name]: - acc = all_results['accuracy_by_method'][method_name][condition['name']]['accuracy'] - row += f" {acc:.1%} " + manifold_outputs = [ + manifold_plot, projection_2d, density_plot, + feature_distributions, correlation_matrix, + species_stats, boundary_stats, similarity_stats + ] + + # Set up event handlers for real-time updates + for component in manifold_inputs: + component.change( + update_manifold_visualization, + inputs=manifold_inputs, + outputs=manifold_outputs + ) + + # Initialize the plots with default values + demo.load( + lambda: update_manifold_visualization( + ["Human", "Dog"], # species_selection + list(df_combined['label'].unique()), # emotion_selection + "gamma", # lens_selection + 0, # alpha_min + 1, # alpha_max + 0, # srl_min + 100, # srl_max + -3, # feature_min + 3, # feature_max + 6, # point_size + True, # show_boundary + False, # show_trajectories + "Species" # color_scheme + ), + outputs=manifold_outputs + ) + + with gr.TabItem("Interactive Holography"): + with gr.Row(): + with gr.Column(scale=1): + gr.Markdown("### Cross-Species Holography Controls") + + # Species selection and automatic pairing + species_dropdown = gr.Dropdown( + label="Select Species", + choices=["Dog", "Human"], + value="Dog" + ) + + # Primary file selection (filtered by species) + dog_files = df_combined[df_combined["source"] == "Dog"]["filepath"].astype(str).tolist() + human_files = df_combined[df_combined["source"] == "Human"]["filepath"].astype(str).tolist() + + primary_dropdown = gr.Dropdown( + label="Primary Audio File", + choices=dog_files, + value=dog_files[0] if dog_files else None + ) + + # Automatically found neighbor (from opposite species) + neighbor_dropdown = gr.Dropdown( + label="Auto-Found Cross-Species Neighbor", + choices=human_files, + value=human_files[0] if human_files else None, + interactive=True # Allow manual override + ) + + holo_lens_dropdown = gr.Dropdown(label="CMT Lens", choices=["gamma", "zeta", "airy", "bessel"], value="gamma") + holo_resolution_slider = gr.Slider(label="Field Resolution", minimum=20, maximum=100, step=5, value=40) + holo_wavelength_slider = gr.Slider(label="Illumination Wavelength (nm)", minimum=380, maximum=750, step=5, value=550) + + # Information panels + primary_info_html = gr.HTML(label="Primary Audio Info") + neighbor_info_html = gr.HTML(label="Neighbor Audio Info") + + # Audio players + primary_audio_out = gr.Audio(label="Primary Audio") + neighbor_audio_out = gr.Audio(label="Neighbor Audio") + + with gr.Column(scale=2): + dual_holography_plot = gr.Plot(label="Side-by-Side Holographic Comparison") + dual_diagnostic_plot = gr.Plot(label="Cross-Species Diagnostic Comparison") + + def update_file_choices(species): + """Update the primary file dropdown based on selected species.""" + species_files = df_combined[df_combined["source"] == species]["filepath"].astype(str).tolist() + return species_files + + def update_cross_species_view(species, primary_file, neighbor_file, lens, resolution, wavelength): + if not primary_file: + empty_fig = go.Figure(layout={"title": "Please select a primary file."}) + return empty_fig, empty_fig, "", "", None, None + + # Get primary row + primary_row = df_combined[ + (df_combined["filepath"] == primary_file) & + (df_combined["source"] == species) + ].iloc[0] if len(df_combined[ + (df_combined["filepath"] == primary_file) & + (df_combined["source"] == species) + ]) > 0 else None + + if primary_row is None: + empty_fig = go.Figure(layout={"title": "Primary file not found."}) + return empty_fig, empty_fig, "", "", None, None, [] + + # Find cross-species neighbor if not manually selected + if not neighbor_file: + neighbor_row = find_nearest_cross_species_neighbor(primary_row, df_combined) + if neighbor_row is not None: + neighbor_file = neighbor_row['filepath'] else: - row += f" {'N/A':<8} " - print(row) - - print("-" * 80) - - # Calculate overall performance metrics - cmt_overall_accuracy = np.mean([ - data['accuracy'] for data in all_results['accuracy_by_method']['CMT_GMT'].values() - ]) - - best_competitor_accuracies = [] - for competitor in competitors: - if competitor in all_results['accuracy_by_method']: - comp_accuracy = np.mean([ - data['accuracy'] for data in all_results['accuracy_by_method'][competitor].values() - ]) - best_competitor_accuracies.append(comp_accuracy) - - best_competitor_accuracy = max(best_competitor_accuracies) if best_competitor_accuracies else 0 - improvement = cmt_overall_accuracy - best_competitor_accuracy - - # GMT-specific metrics - avg_gmt_dimensions = np.mean([ - data['avg_dimensions'] for data in all_results['accuracy_by_method']['CMT_GMT'].values() - if 'avg_dimensions' in data - ]) - - avg_gmt_confidence = np.mean([ - data['avg_confidence'] for data in all_results['accuracy_by_method']['CMT_GMT'].values() - if 'avg_confidence' in data - ]) - - print(f"\nπ FINAL COMPREHENSIVE RESULTS") - print("=" * 50) - print(f"β CMT-GMT Overall Accuracy: {cmt_overall_accuracy:.1%}") - print(f"π Best Competitor Accuracy: {best_competitor_accuracy:.1%}") - print(f"π CMT Improvement: +{improvement:.1%} ({improvement*100:.1f} percentage points)") - print(f"π¬ Average GMT Dimensions: {avg_gmt_dimensions:.1f}") - print(f"π― Average GMT Confidence: {avg_gmt_confidence:.3f}") - print(f"π Mathematical Lenses Used: {cmt_engine.n_lenses}") - print(f"π Multi-view Architecture: {cmt_engine.n_views} views") - - # Statistical significance - if improvement > 0.02: # 2 percentage point threshold - print(f"π Statistical Significance: CONFIRMED (>{improvement*100:.1f}pp improvement)") - else: - print(f"π Statistical Significance: MARGINAL (<2pp improvement)") - - print(f"\nπ‘ REVOLUTIONARY GMT BREAKTHROUGH CONFIRMED") - print("=" * 50) - print(f"β’ Pure GMT mathematics achieves {cmt_overall_accuracy:.1%} accuracy") - print(f"β’ {avg_gmt_dimensions:.0f}+ dimensional feature space from mathematical lenses") - print(f"β’ NO FFT/wavelets/DTF preprocessing required") - print(f"β’ Robust performance under extreme aerospace conditions") - print(f"β’ Multi-lens architecture enables comprehensive fault signatures") - print(f"β’ Ready for immediate commercial deployment") - - return { - 'cmt_overall_accuracy': cmt_overall_accuracy, - 'best_competitor_accuracy': best_competitor_accuracy, - 'improvement_percentage': improvement * 100, - 'avg_gmt_dimensions': avg_gmt_dimensions, - 'avg_gmt_confidence': avg_gmt_confidence, - 'statistical_significance': improvement > 0.02, - 'test_conditions': len(test_conditions), - 'total_samples': len(fault_types) * len(test_conditions) * 10, # samples tested - 'all_results': all_results - } + # Get manually selected neighbor + opposite_species = 'Human' if species == 'Dog' else 'Dog' + neighbor_row = df_combined[ + (df_combined["filepath"] == neighbor_file) & + (df_combined["source"] == opposite_species) + ].iloc[0] if len(df_combined[ + (df_combined["filepath"] == neighbor_file) & + (df_combined["source"] == opposite_species) + ]) > 0 else None + + # Get CMT data directly from CSV (no audio processing needed!) + print(f"π Using preprocessed CMT data for: {primary_row['filepath']} ({lens} lens)") + primary_cmt = get_cmt_data_from_csv(primary_row, lens) + + neighbor_cmt = None + if neighbor_row is not None: + print(f"π Using preprocessed CMT data for: {neighbor_row['filepath']} ({lens} lens)") + neighbor_cmt = get_cmt_data_from_csv(neighbor_row, lens) + + # Get audio file paths only for playback + primary_fp = resolve_audio_path(primary_row) + neighbor_fp = resolve_audio_path(neighbor_row) if neighbor_row is not None else None + + # Create visualizations + if primary_cmt and neighbor_cmt: + primary_title = f"{species}: {primary_row.get('label', 'Unknown')}" + neighbor_title = f"{neighbor_row['source']}: {neighbor_row.get('label', 'Unknown')}" + + dual_holo_fig = create_dual_holography_plot( + primary_cmt["z"], primary_cmt["phi"], + neighbor_cmt["z"], neighbor_cmt["phi"], + resolution, wavelength, primary_title, neighbor_title + ) + + dual_diag_fig = create_dual_diagnostic_plots( + primary_cmt["z"], primary_cmt["w"], + neighbor_cmt["z"], neighbor_cmt["w"], + primary_title, neighbor_title + ) + else: + dual_holo_fig = go.Figure(layout={"title": "Error processing audio files"}) + dual_diag_fig = go.Figure(layout={"title": "Error processing audio files"}) + + # Build info strings with CMT diagnostic values + primary_info = f""" + Primary: {primary_row['filepath']}