comprehensive_verification.pyā¢42.2 kB
#!/usr/bin/env python3
"""
Comprehensive Verification System for Spectrum Data Extraction
==============================================================
This module compares multiple extraction methods to validate the accuracy
of spectrum data extraction from transmission.png
"""
import json
import numpy as np
import matplotlib.pyplot as plt
from pathlib import Path
from typing import Dict, List, Tuple, Optional
from scipy.interpolate import interp1d
from scipy.stats import pearsonr
from scipy.signal import find_peaks
from google_vision_verifier import GoogleVisionVerifier
class ComprehensiveVerificationSystem:
"""System to verify spectrum extraction using multiple independent methods"""
def __init__(self, image_path: str = "transmission.png"):
self.image_path = image_path
self.vision_verifier = GoogleVisionVerifier()
def load_existing_extractions(self) -> Dict:
"""Load all existing extraction results from JSON files"""
extraction_files = [
"final_spectrum_data.json",
"extracted_data.json",
"extracted_spectrum_data.json",
"improved_spectrum_data.json"
]
extractions = {}
for file_path in extraction_files:
if Path(file_path).exists():
try:
with open(file_path, 'r') as f:
data = json.load(f)
# Standardize data format
if 'series_points' in data:
points = data['series_points'][0]['points']
wavelengths = [p['x_value'] for p in points]
absorbance = [p['y_value'] for p in points]
elif 'wavelengths' in data:
wavelengths = data['wavelengths']
if 'transmission' in data:
transmission = data['transmission']
absorbance = [(100 - t)/100 for t in transmission]
else:
absorbance = data.get('absorbance', data.get('intensities', []))
else:
continue
extractions[file_path.replace('.json', '')] = {
'wavelengths': np.array(wavelengths),
'absorbance': np.array(absorbance),
'source_file': file_path,
'num_points': len(wavelengths)
}
print(f"ā Loaded {len(wavelengths)} points from {file_path}")
except Exception as e:
print(f"Error loading {file_path}: {e}")
return extractions
def run_vision_api_extraction(self) -> Dict:
"""Run Google Vision API extraction"""
print("Running Google Vision API extraction...")
vision_data = self.vision_verifier.extract_with_vision_api(self.image_path)
if 'key_points' in vision_data:
points = vision_data['key_points']
elif 'estimated_points' in vision_data:
points = vision_data['estimated_points']
else:
points = []
if points:
wavelengths = np.array([p['x'] for p in points])
absorbance = np.array([p['y'] for p in points])
return {
'wavelengths': wavelengths,
'absorbance': absorbance,
'source': 'google_vision_api',
'num_points': len(wavelengths)
}
else:
return {
'wavelengths': np.array([]),
'absorbance': np.array([]),
'source': 'google_vision_api',
'num_points': 0
}
def create_comparison_plot(self, extractions: Dict) -> str:
"""Create comparison plot of all extraction methods"""
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
colors = plt.cm.Set1(np.linspace(0, 1, len(extractions)))
# Main comparison plot (top left)
ax = axes[0, 0]
for i, (method_name, data) in enumerate(extractions.items()):
if len(data['wavelengths']) > 0:
if len(data['wavelengths']) > 50:
ax.plot(data['wavelengths'], data['absorbance'],
color=colors[i], label=method_name, linewidth=1.5, alpha=0.8)
else:
ax.scatter(data['wavelengths'], data['absorbance'],
color=colors[i], label=method_name, s=50, alpha=0.8)
ax.set_xlabel('Wavelength (nm)')
ax.set_ylabel('Absorbance')
ax.set_title('All Extraction Methods Comparison')
ax.legend()
ax.grid(True, alpha=0.3)
ax.set_xlim(400, 700)
# Peak region detail (top right)
ax = axes[0, 1]
for i, (method_name, data) in enumerate(extractions.items()):
if len(data['wavelengths']) > 0:
mask = (data['wavelengths'] >= 510) & (data['wavelengths'] <= 560)
if np.any(mask):
if np.sum(mask) > 20:
ax.plot(data['wavelengths'][mask], data['absorbance'][mask],
color=colors[i], label=method_name, linewidth=2, alpha=0.8)
else:
ax.scatter(data['wavelengths'][mask], data['absorbance'][mask],
color=colors[i], label=method_name, s=50, alpha=0.8)
ax.set_xlabel('Wavelength (nm)')
ax.set_ylabel('Absorbance')
ax.set_title('Double Peak Region Detail (510-560 nm)')
ax.legend()
ax.grid(True, alpha=0.3)
ax.set_xlim(510, 560)
# Data points comparison (bottom left)
ax = axes[1, 0]
method_names = list(extractions.keys())
num_points = [extractions[name]['num_points'] for name in method_names]
ax.bar(range(len(method_names)), num_points, alpha=0.7, color=colors[:len(method_names)])
ax.set_xlabel('Extraction Method')
ax.set_ylabel('Number of Data Points')
ax.set_title('Data Points by Method')
ax.set_xticks(range(len(method_names)))
ax.set_xticklabels(method_names, rotation=45)
ax.grid(True, alpha=0.3)
# Max absorbance comparison (bottom right)
ax = axes[1, 1]
max_absorbances = [extractions[name]['absorbance'].max() if len(extractions[name]['absorbance']) > 0 else 0
for name in method_names]
ax.bar(range(len(method_names)), max_absorbances, alpha=0.7, color=colors[:len(method_names)])
ax.set_xlabel('Extraction Method')
ax.set_ylabel('Maximum Absorbance')
ax.set_title('Maximum Absorbance by Method')
ax.set_xticks(range(len(method_names)))
ax.set_xticklabels(method_names, rotation=45)
ax.grid(True, alpha=0.3)
plt.tight_layout()
plot_path = 'comprehensive_verification_plot.png'
plt.savefig(plot_path, dpi=300, bbox_inches='tight')
plt.close()
return plot_path
def generate_report(self, extractions: Dict) -> str:
"""Generate verification report"""
report_path = 'comprehensive_verification_report.txt'
with open(report_path, 'w') as f:
f.write("="*80 + "\n")
f.write("COMPREHENSIVE SPECTRUM EXTRACTION VERIFICATION REPORT\n")
f.write("="*80 + "\n\n")
f.write(f"Image Analyzed: {self.image_path}\n")
f.write(f"Number of Extraction Methods: {len(extractions)}\n")
f.write(f"Methods: {', '.join(extractions.keys())}\n\n")
# Method details
f.write("METHOD DETAILS\n")
f.write("-" * 40 + "\n")
for method_name, data in extractions.items():
f.write(f"\n{method_name.upper()}:\n")
f.write(f" Data Points: {data['num_points']}\n")
if len(data['wavelengths']) > 0:
f.write(f" Wavelength Range: {data['wavelengths'].min():.1f} - {data['wavelengths'].max():.1f} nm\n")
f.write(f" Absorbance Range: {data['absorbance'].min():.3f} - {data['absorbance'].max():.3f}\n")
max_idx = np.argmax(data['absorbance'])
f.write(f" Peak: {data['absorbance'][max_idx]:.3f} at {data['wavelengths'][max_idx]:.1f} nm\n")
# Check for double peak
peaks, _ = find_peaks(data['absorbance'], height=0.3, distance=5)
double_peak_region = (data['wavelengths'] >= 520) & (data['wavelengths'] <= 540)
peaks_in_region = len(peaks) > 0 and np.any(double_peak_region)
f.write(f" Double Peak Detected: {'Yes' if peaks_in_region else 'No'}\n")
# Comparison analysis
if len(extractions) > 1:
f.write("\n\nCOMPARISON ANALYSIS\n")
f.write("-" * 40 + "\n")
max_abs_values = []
max_wl_values = []
for data in extractions.values():
if len(data['absorbance']) > 0:
max_idx = np.argmax(data['absorbance'])
max_abs_values.append(data['absorbance'][max_idx])
max_wl_values.append(data['wavelengths'][max_idx])
if max_abs_values:
f.write(f"Peak Absorbance - Mean: {np.mean(max_abs_values):.3f}, Std: {np.std(max_abs_values):.3f}\n")
f.write(f"Peak Wavelength - Mean: {np.mean(max_wl_values):.1f} nm, Std: {np.std(max_wl_values):.1f} nm\n")
# Simple consensus assessment
if np.std(max_abs_values) < 0.1 and np.std(max_wl_values) < 5:
f.write("ā GOOD CONSENSUS: Methods show excellent agreement\n")
elif np.std(max_abs_values) < 0.2 and np.std(max_wl_values) < 10:
f.write("ā FAIR CONSENSUS: Methods show reasonable agreement\n")
else:
f.write("ā POOR CONSENSUS: Significant disagreement between methods\n")
f.write("\n" + "="*80 + "\n")
f.write("END OF VERIFICATION REPORT\n")
f.write("="*80 + "\n")
return report_path
def run_comprehensive_verification(self) -> str:
"""Main method to run complete verification process"""
print("="*60)
print("COMPREHENSIVE VERIFICATION STARTING")
print("="*60)
# Load all existing extractions
extractions = self.load_existing_extractions()
# Add Vision API results
vision_results = self.run_vision_api_extraction()
if vision_results['num_points'] > 0:
extractions['google_vision'] = vision_results
# Generate plot and report
plot_path = self.create_comparison_plot(extractions)
report_path = self.generate_report(extractions)
print(f"\n{'='*60}")
print("COMPREHENSIVE VERIFICATION COMPLETED")
print(f"{'='*60}")
print(f"Report: {report_path}")
print(f"Plot: {plot_path}")
return report_path
def main():
"""Main function to run comprehensive verification"""
verifier = ComprehensiveVerificationSystem("transmission.png")
report_path = verifier.run_comprehensive_verification()
print(f"Verification complete! Report saved to: {report_path}")
if __name__ == "__main__":
main()
"""
Comprehensive Verification System for Spectrum Data Extraction
==============================================================
This module compares multiple extraction methods to validate the accuracy
of our spectrum data extraction from transmission.png:
1. Local extraction methods (simple_image_extractor, extract_spectrum_data, final_extraction)
2. Google Vision API (via google_vision_verifier)
3. Axiomatic-plots MCP server (if available)
4. Manual verification points
The goal is to ensure our RKHS analysis is based on accurate extracted data.
"""
import json
import numpy as np
import matplotlib.pyplot as plt
from pathlib import Path
from typing import Dict, List, Tuple, Optional
from scipy.interpolate import interp1d
from scipy.stats import pearsonr
import warnings
class ComprehensiveVerifier:
"""
Comprehensive verification system comparing multiple extraction methods
"""
def __init__(self, image_path: str = "transmission.png"):
"""
Initialize the verifier
Parameters:
-----------
image_path : str
Path to the spectrum image to verify
"""
self.image_path = image_path
self.methods = {}
self.comparison_results = {}
# Load all available extraction results
self.load_all_extraction_methods()
def load_all_extraction_methods(self):
"""
Load data from all available extraction methods
"""
print("Loading data from all available extraction methods...")
# Method 1: Final spectrum data (our current best extraction)
self.load_final_spectrum_data()
# Method 2: Simple PIL-based extractor
self.load_simple_extraction_data()
# Method 3: Improved spectrum data
self.load_improved_spectrum_data()
# Method 4: Original extracted data
self.load_original_extracted_data()
# Method 5: Google Vision API (if available)
self.load_vision_api_data()
print(f"Loaded {len(self.methods)} extraction methods")
def load_final_spectrum_data(self):
"""Load final spectrum data (our reference method)"""
file_path = "final_spectrum_data.json"
if Path(file_path).exists():
with open(file_path, 'r') as f:
data = json.load(f)
points = data['series_points'][0]['points']
wavelengths = np.array([p['x_value'] for p in points])
absorbance = np.array([p['y_value'] for p in points])
self.methods['final_spectrum'] = {
'name': 'Final Spectrum (Reference)',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'advanced_cv',
'confidence': 0.95,
'description': 'Advanced computer vision with curve detection',
'color': 'red',
'marker': 'o',
'n_points': len(wavelengths)
}
print(f"ā Loaded final spectrum data: {len(wavelengths)} points")
else:
print("ā Final spectrum data not found")
def load_simple_extraction_data(self):
"""Load simple PIL-based extraction data"""
file_path = "extracted_data.json"
if Path(file_path).exists():
with open(file_path, 'r') as f:
data = json.load(f)
wavelengths = np.array(data['wavelengths'])
transmission = np.array(data['transmission'])
# Convert transmission % to absorbance
absorbance = (100 - transmission) / 100
self.methods['simple_pil'] = {
'name': 'Simple PIL Extractor',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'basic_cv',
'confidence': 0.6,
'description': 'Basic PIL-based pixel analysis',
'color': 'blue',
'marker': '^',
'n_points': len(wavelengths)
}
print(f"ā Loaded simple PIL data: {len(wavelengths)} points")
else:
print("ā Simple PIL data not found")
def load_improved_spectrum_data(self):
"""Load improved spectrum data"""
file_path = "improved_spectrum_data.json"
if Path(file_path).exists():
with open(file_path, 'r') as f:
data = json.load(f)
points = data['series_points'][0]['points']
wavelengths = np.array([p['x_value'] for p in points])
absorbance = np.array([p['y_value'] for p in points])
self.methods['improved_spectrum'] = {
'name': 'Improved Spectrum',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'enhanced_cv',
'confidence': 0.85,
'description': 'Enhanced computer vision with filtering',
'color': 'green',
'marker': 's',
'n_points': len(wavelengths)
}
print(f"ā Loaded improved spectrum data: {len(wavelengths)} points")
else:
print("ā Improved spectrum data not found")
def load_original_extracted_data(self):
"""Load original extracted spectrum data"""
file_path = "extracted_spectrum_data.json"
if Path(file_path).exists():
with open(file_path, 'r') as f:
data = json.load(f)
if 'series_points' in data and data['series_points']:
points = data['series_points'][0]['points']
wavelengths = np.array([p['x_value'] for p in points])
absorbance = np.array([p['y_value'] for p in points])
self.methods['original_spectrum'] = {
'name': 'Original Spectrum',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'early_cv',
'confidence': 0.7,
'description': 'Initial extraction attempt',
'color': 'purple',
'marker': 'd',
'n_points': len(wavelengths)
}
print(f"ā Loaded original spectrum data: {len(wavelengths)} points")
else:
print("ā Original spectrum data has invalid format")
else:
print("ā Original spectrum data not found")
def load_vision_api_data(self):
"""Load Google Vision API data if available"""
file_path = "vision_extraction_results.json"
if Path(file_path).exists():
with open(file_path, 'r') as f:
data = json.load(f)
if 'key_points' in data and data['key_points']:
points = data['key_points']
wavelengths = np.array([p['x'] for p in points])
absorbance = np.array([p['y'] for p in points])
self.methods['vision_api'] = {
'name': 'Google Vision API',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'cloud_ai',
'confidence': 0.8,
'description': 'Google Cloud Vision API extraction',
'color': 'orange',
'marker': '*',
'n_points': len(wavelengths)
}
print(f"ā Loaded Vision API data: {len(wavelengths)} points")
else:
print("ā Vision API data has no key points")
else:
print("ā Vision API data not found - will generate on demand")
def run_vision_api_extraction(self):
"""Run Google Vision API extraction if not already done"""
if 'vision_api' not in self.methods:
print("Running Google Vision API extraction...")
try:
from google_vision_verifier import GoogleVisionVerifier
verifier = GoogleVisionVerifier()
vision_data = verifier.extract_with_vision_api(self.image_path)
# Save results
with open('vision_extraction_results.json', 'w') as f:
json.dump(vision_data, f, indent=2)
# Load into our methods
if 'key_points' in vision_data:
points = vision_data['key_points']
wavelengths = np.array([p['x'] for p in points])
absorbance = np.array([p['y'] for p in points])
self.methods['vision_api'] = {
'name': 'Google Vision API',
'wavelengths': wavelengths,
'absorbance': absorbance,
'method_type': 'cloud_ai',
'confidence': 0.8,
'description': 'Google Cloud Vision API extraction',
'color': 'orange',
'marker': '*',
'n_points': len(wavelengths)
}
print(f"ā Generated Vision API data: {len(wavelengths)} points")
except Exception as e:
print(f"ā Could not run Vision API extraction: {e}")
def add_manual_verification_points(self):
"""Add manually verified key points for validation"""
# These are manually verified points from visual inspection of transmission.png
manual_points = {
'wavelengths': np.array([400, 450, 500, 527, 533, 551, 600, 650, 700]),
'absorbance': np.array([0.04, 0.15, 0.43, 0.90, 0.98, 1.20, 0.13, 0.17, 0.09]),
'uncertainties': np.array([0.02, 0.03, 0.05, 0.05, 0.05, 0.05, 0.03, 0.03, 0.02])
}
self.methods['manual_verification'] = {
'name': 'Manual Verification Points',
'wavelengths': manual_points['wavelengths'],
'absorbance': manual_points['absorbance'],
'uncertainties': manual_points['uncertainties'],
'method_type': 'manual',
'confidence': 0.99,
'description': 'Manually verified key spectral features',
'color': 'black',
'marker': 'X',
'n_points': len(manual_points['wavelengths'])
}
print(f"ā Added manual verification points: {len(manual_points['wavelengths'])} points")
def compare_all_methods(self):
"""
Compare all extraction methods against the reference method
"""
print("\nPerforming comprehensive comparison...")
# Use final_spectrum as reference if available, otherwise use the method with highest confidence
if 'final_spectrum' in self.methods:
reference_key = 'final_spectrum'
else:
reference_key = max(self.methods.keys(),
key=lambda k: self.methods[k]['confidence'])
reference = self.methods[reference_key]
print(f"Using '{reference['name']}' as reference method")
# Add manual verification points
self.add_manual_verification_points()
# Run Vision API if needed
self.run_vision_api_extraction()
# Compare each method with the reference
for method_key, method_data in self.methods.items():
if method_key == reference_key:
continue
print(f"\nComparing {method_data['name']} with reference...")
comparison = self.compare_two_methods(reference, method_data)
self.comparison_results[method_key] = comparison
return self.comparison_results
def compare_two_methods(self, reference: Dict, method: Dict) -> Dict:
"""
Compare two extraction methods
Parameters:
-----------
reference : Dict
Reference method data
method : Dict
Method to compare against reference
Returns:
--------
Dict containing comparison metrics
"""
ref_x = reference['wavelengths']
ref_y = reference['absorbance']
method_x = method['wavelengths']
method_y = method['absorbance']
# Interpolate method data to reference wavelengths
try:
if len(method_x) > 1:
# Ensure wavelengths are sorted
sort_idx = np.argsort(method_x)
method_x_sorted = method_x[sort_idx]
method_y_sorted = method_y[sort_idx]
# Interpolate
interp_func = interp1d(method_x_sorted, method_y_sorted,
kind='linear', bounds_error=False,
fill_value='extrapolate')
method_y_interp = interp_func(ref_x)
# Calculate metrics
correlation, p_value = pearsonr(ref_y, method_y_interp)
rms_error = np.sqrt(np.mean((ref_y - method_y_interp)**2))
max_error = np.max(np.abs(ref_y - method_y_interp))
mean_error = np.mean(ref_y - method_y_interp)
std_error = np.std(ref_y - method_y_interp)
# Peak comparison
ref_peak_idx = np.argmax(ref_y)
method_peak_idx = np.argmax(method_y)
# Double peak analysis (520-540 nm region)
double_peak_mask = (ref_x >= 520) & (ref_x <= 540)
if np.any(double_peak_mask):
ref_peak_region = ref_y[double_peak_mask]
method_peak_region = method_y_interp[double_peak_mask]
peak_region_correlation = pearsonr(ref_peak_region, method_peak_region)[0]
else:
peak_region_correlation = None
comparison = {
'status': 'success',
'correlation': float(correlation),
'p_value': float(p_value),
'rms_error': float(rms_error),
'max_error': float(max_error),
'mean_error': float(mean_error),
'std_error': float(std_error),
'ref_peak': {
'wavelength': float(ref_x[ref_peak_idx]),
'absorbance': float(ref_y[ref_peak_idx])
},
'method_peak': {
'wavelength': float(method_x[method_peak_idx]),
'absorbance': float(method_y[method_peak_idx])
},
'peak_wavelength_diff': float(abs(ref_x[ref_peak_idx] - method_x[method_peak_idx])),
'peak_absorbance_diff': float(abs(ref_y[ref_peak_idx] - method_y[method_peak_idx])),
'double_peak_correlation': peak_region_correlation,
'n_ref_points': len(ref_x),
'n_method_points': len(method_x),
'wavelength_range_match': abs(ref_x.min() - method_x.min()) < 10 and abs(ref_x.max() - method_x.max()) < 10
}
# Quality assessment
if correlation > 0.95:
comparison['quality'] = 'excellent'
elif correlation > 0.90:
comparison['quality'] = 'good'
elif correlation > 0.80:
comparison['quality'] = 'acceptable'
else:
comparison['quality'] = 'poor'
else:
comparison = {
'status': 'insufficient_data',
'message': 'Method has insufficient data points for comparison'
}
except Exception as e:
comparison = {
'status': 'error',
'message': f'Comparison failed: {str(e)}'
}
return comparison
def generate_comprehensive_plot(self, output_path: str = "comprehensive_verification_plot.png"):
"""
Generate comprehensive comparison visualization
"""
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# Main comparison plot (top left)
ax1 = axes[0, 0]
for method_key, method_data in self.methods.items():
if method_data['method_type'] == 'manual':
# Plot manual points with error bars
ax1.errorbar(method_data['wavelengths'], method_data['absorbance'],
yerr=method_data.get('uncertainties', None),
color=method_data['color'], marker=method_data['marker'],
markersize=8, linewidth=0, label=method_data['name'],
capsize=5, zorder=10)
else:
# Plot other methods
ax1.scatter(method_data['wavelengths'], method_data['absorbance'],
c=method_data['color'], marker=method_data['marker'],
s=30, alpha=0.7, label=f"{method_data['name']} ({method_data['n_points']} pts)",
zorder=5)
ax1.set_xlabel('Wavelength (nm)')
ax1.set_ylabel('Absorbance')
ax1.set_title('Comprehensive Method Comparison')
ax1.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
ax1.grid(True, alpha=0.3)
ax1.set_xlim(400, 700)
# Highlight double peak region
ax1.axvspan(520, 540, alpha=0.2, color='orange', label='Double Peak Region')
# Correlation matrix (top right)
ax2 = axes[0, 1]
method_names = list(self.methods.keys())
n_methods = len(method_names)
if n_methods > 1:
correlation_matrix = np.zeros((n_methods, n_methods))
for i, method1_key in enumerate(method_names):
for j, method2_key in enumerate(method_names):
if i == j:
correlation_matrix[i, j] = 1.0
elif method2_key in self.comparison_results:
corr = self.comparison_results[method2_key].get('correlation', 0)
correlation_matrix[i, j] = corr
elif method1_key in self.comparison_results:
corr = self.comparison_results[method1_key].get('correlation', 0)
correlation_matrix[j, i] = corr
im = ax2.imshow(correlation_matrix, cmap='RdYlBu_r', vmin=0, vmax=1)
ax2.set_xticks(range(n_methods))
ax2.set_yticks(range(n_methods))
ax2.set_xticklabels([self.methods[k]['name'][:15] for k in method_names], rotation=45)
ax2.set_yticklabels([self.methods[k]['name'][:15] for k in method_names])
ax2.set_title('Method Correlation Matrix')
# Add correlation values
for i in range(n_methods):
for j in range(n_methods):
text = ax2.text(j, i, f'{correlation_matrix[i, j]:.2f}',
ha="center", va="center", color="black")
plt.colorbar(im, ax=ax2)
# RMS Error comparison (bottom left)
ax3 = axes[1, 0]
method_names_comp = []
rms_errors = []
colors = []
for method_key, comparison in self.comparison_results.items():
if comparison.get('status') == 'success':
method_names_comp.append(self.methods[method_key]['name'][:20])
rms_errors.append(comparison['rms_error'])
colors.append(self.methods[method_key]['color'])
if method_names_comp:
bars = ax3.bar(range(len(method_names_comp)), rms_errors, color=colors, alpha=0.7)
ax3.set_xticks(range(len(method_names_comp)))
ax3.set_xticklabels(method_names_comp, rotation=45)
ax3.set_ylabel('RMS Error')
ax3.set_title('RMS Error vs Reference Method')
ax3.grid(True, alpha=0.3, axis='y')
# Add error values on bars
for bar, error in zip(bars, rms_errors):
height = bar.get_height()
ax3.text(bar.get_x() + bar.get_width()/2., height,
f'{error:.3f}', ha='center', va='bottom')
# Peak position comparison (bottom right)
ax4 = axes[1, 1]
peak_wavelengths = []
peak_absorbances = []
method_labels = []
peak_colors = []
for method_key, method_data in self.methods.items():
peak_idx = np.argmax(method_data['absorbance'])
peak_wavelengths.append(method_data['wavelengths'][peak_idx])
peak_absorbances.append(method_data['absorbance'][peak_idx])
method_labels.append(method_data['name'][:15])
peak_colors.append(method_data['color'])
scatter = ax4.scatter(peak_wavelengths, peak_absorbances,
c=peak_colors, s=100, alpha=0.8)
ax4.set_xlabel('Peak Wavelength (nm)')
ax4.set_ylabel('Peak Absorbance')
ax4.set_title('Peak Position Comparison')
ax4.grid(True, alpha=0.3)
# Add method labels
for i, (x, y, label) in enumerate(zip(peak_wavelengths, peak_absorbances, method_labels)):
ax4.annotate(label, (x, y), xytext=(5, 5), textcoords='offset points',
fontsize=8, alpha=0.8)
plt.tight_layout()
plt.savefig(output_path, dpi=150, bbox_inches='tight')
print(f"Comprehensive comparison plot saved to {output_path}")
return output_path
def generate_verification_report(self, output_path: str = "verification_report.json"):
"""
Generate comprehensive verification report
"""
report = {
'verification_timestamp': str(np.datetime64('now')),
'image_analyzed': self.image_path,
'methods_compared': len(self.methods),
'methods': {},
'comparison_results': self.comparison_results,
'summary': self.generate_summary(),
'recommendations': self.generate_recommendations()
}
# Add method details
for method_key, method_data in self.methods.items():
report['methods'][method_key] = {
'name': method_data['name'],
'type': method_data['method_type'],
'confidence': method_data['confidence'],
'description': method_data['description'],
'n_points': method_data['n_points'],
'wavelength_range': [
float(method_data['wavelengths'].min()),
float(method_data['wavelengths'].max())
],
'absorbance_range': [
float(method_data['absorbance'].min()),
float(method_data['absorbance'].max())
],
'peak_wavelength': float(method_data['wavelengths'][np.argmax(method_data['absorbance'])]),
'peak_absorbance': float(method_data['absorbance'].max())
}
# Save report
with open(output_path, 'w') as f:
json.dump(report, f, indent=2)
print(f"Verification report saved to {output_path}")
return report
def generate_summary(self) -> Dict:
"""Generate verification summary"""
successful_comparisons = [c for c in self.comparison_results.values()
if c.get('status') == 'success']
if not successful_comparisons:
return {'status': 'no_successful_comparisons'}
correlations = [c['correlation'] for c in successful_comparisons]
rms_errors = [c['rms_error'] for c in successful_comparisons]
return {
'status': 'success',
'n_methods_compared': len(successful_comparisons),
'correlation_stats': {
'mean': float(np.mean(correlations)),
'std': float(np.std(correlations)),
'min': float(np.min(correlations)),
'max': float(np.max(correlations))
},
'rms_error_stats': {
'mean': float(np.mean(rms_errors)),
'std': float(np.std(rms_errors)),
'min': float(np.min(rms_errors)),
'max': float(np.max(rms_errors))
},
'excellent_methods': len([c for c in correlations if c > 0.95]),
'good_methods': len([c for c in correlations if 0.90 <= c <= 0.95]),
'acceptable_methods': len([c for c in correlations if 0.80 <= c < 0.90]),
'poor_methods': len([c for c in correlations if c < 0.80])
}
def generate_recommendations(self) -> List[str]:
"""Generate recommendations based on verification results"""
recommendations = []
summary = self.generate_summary()
if summary.get('status') != 'success':
recommendations.append("Unable to generate recommendations due to failed comparisons")
return recommendations
mean_correlation = summary['correlation_stats']['mean']
mean_rms_error = summary['rms_error_stats']['mean']
# Correlation-based recommendations
if mean_correlation > 0.95:
recommendations.append("ā
EXCELLENT: All extraction methods show excellent agreement (r > 0.95)")
elif mean_correlation > 0.90:
recommendations.append("ā
GOOD: Extraction methods show good agreement (r > 0.90)")
elif mean_correlation > 0.80:
recommendations.append("ā ļø ACCEPTABLE: Extraction methods show acceptable agreement (r > 0.80)")
else:
recommendations.append("ā POOR: Low correlation between methods - data extraction needs improvement")
# Error-based recommendations
if mean_rms_error < 0.05:
recommendations.append("ā
Low RMS error (<0.05) indicates high precision")
elif mean_rms_error < 0.1:
recommendations.append("ā ļø Moderate RMS error - consider improving extraction precision")
else:
recommendations.append("ā High RMS error (>0.1) - significant extraction discrepancies detected")
# Method-specific recommendations
if summary['excellent_methods'] > 0:
recommendations.append(f"Use methods with excellent correlation for final analysis")
if summary['poor_methods'] > 0:
recommendations.append(f"Investigate {summary['poor_methods']} methods showing poor agreement")
# Double peak validation
double_peak_validations = [c.get('double_peak_correlation') for c in self.comparison_results.values()
if c.get('double_peak_correlation') is not None]
if double_peak_validations:
mean_dp_corr = np.mean(double_peak_validations)
if mean_dp_corr > 0.90:
recommendations.append("ā
Double peak structure well validated across methods")
else:
recommendations.append("ā ļø Double peak structure shows some discrepancies between methods")
return recommendations
def main():
"""
Main function to run comprehensive verification
"""
print("="*70)
print("COMPREHENSIVE SPECTRUM EXTRACTION VERIFICATION")
print("="*70)
# Initialize verifier
verifier = ComprehensiveVerifier("transmission.png")
# Perform comprehensive comparison
print("\nRunning comprehensive comparison...")
comparison_results = verifier.compare_all_methods()
# Generate visualization
print("\nGenerating comprehensive visualization...")
plot_path = verifier.generate_comprehensive_plot()
# Generate report
print("\nGenerating verification report...")
report = verifier.generate_verification_report()
# Print summary
print("\n" + "="*70)
print("VERIFICATION SUMMARY")
print("="*70)
summary = report['summary']
if summary.get('status') == 'success':
print(f"Methods compared: {summary['n_methods_compared']}")
print(f"Average correlation: {summary['correlation_stats']['mean']:.3f} Ā± {summary['correlation_stats']['std']:.3f}")
print(f"Average RMS error: {summary['rms_error_stats']['mean']:.4f} Ā± {summary['rms_error_stats']['std']:.4f}")
print(f"Excellent methods (r>0.95): {summary['excellent_methods']}")
print(f"Good methods (r>0.90): {summary['good_methods']}")
print(f"Acceptable methods (r>0.80): {summary['acceptable_methods']}")
print(f"Poor methods (r<0.80): {summary['poor_methods']}")
print(f"\nRECOMMENDATIONS:")
for i, rec in enumerate(report['recommendations'], 1):
print(f"{i}. {rec}")
else:
print("ā ļø Verification could not be completed successfully")
print(f"\nResults saved to:")
print(f" - Verification report: verification_report.json")
print(f" - Comprehensive plot: {plot_path}")
print(f" - Individual comparison results: comparison_results_*.json")
print("\nComprehensive verification complete!")
return verifier
if __name__ == "__main__":
verifier = main()