""" DNA-stabilized Silver Nanocluster Mass Spectrometry Analysis Web Application This Flask-based web application analyzes mass spectrometry data for DNA-silver nanoclusters, providing isotope pattern overlay, charge state identification, and composition analysis. """ from __future__ import annotations import logging import os import sys from typing import Any, Callable, TypeVar import numpy as np import numpy.typing as npt from flask import Flask, Response, jsonify, render_template, request, send_file # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) # Type aliases (guarded for Python < 3.9 / older NumPy) try: NDArrayFloat = npt.NDArray[np.floating[Any]] NDArrayInt = npt.NDArray[np.integer[Any]] except TypeError: NDArrayFloat = np.ndarray # type: ignore[misc] NDArrayInt = np.ndarray # type: ignore[misc] F = TypeVar('F', bound=Callable[..., Any]) # Flask response type (can be Response, tuple of Response/dict and status code, or str) try: FlaskResponse = Response | tuple[Response, int] | str except TypeError: FlaskResponse = Any # type: ignore[misc] # Add lib directory to path for local imports current_dir = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(current_dir, 'lib')) from core.analyzer import MAX_SILVER, MAX_STRANDS, DNASilverAnalyzer # Add current directory to path for local imports sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # Check IsoSpecPy availability for faster isotope pattern generation import importlib.util from pythoms.molecule import composition_from_formula from pythoms.senko_charge_assignment import detect_all_peaks_with_charge ISOSPEC_AVAILABLE = importlib.util.find_spec('IsoSpecPy') is not None if ISOSPEC_AVAILABLE: logger.info('IsoSpecPy available - faster isotope pattern generation enabled') else: logger.warning('IsoSpecPy not installed - using PythoMS (pip install IsoSpecPy for faster performance)') # Toggle for isotope pattern library: 'isospec' (faster) or 'pythoms' (original) # Set to 'isospec' if available, otherwise fall back to 'pythoms' ISOTOPE_LIBRARY = 'isospec' if ISOSPEC_AVAILABLE else 'pythoms' DEFAULT_RESOLUTION = 20000 PEAK_WINDOW = 3.0 def parse_adduct_items(adducts_input: list[dict], adduct_library: dict) -> tuple[float, int, str, dict[str, int]]: """Parse adduct input list and return total mass, charge, display string, and element composition.""" total_adduct_mass = 0.0 total_adduct_charge = 0 adduct_formula_parts: list[str] = [] adduct_elements: dict[str, int] = {} for adduct_entry in adducts_input: adduct_name = adduct_entry.get('name') adduct_count = int(adduct_entry.get('count', 1)) inline_mass = adduct_entry.get('mass') inline_charge = adduct_entry.get('charge') if inline_mass is not None and inline_charge is not None: adduct_mass = float(inline_mass) adduct_charge = int(inline_charge) elif adduct_name in adduct_library: adduct_mass, adduct_charge = adduct_library[adduct_name] else: logger.warning(f"Adduct '{adduct_name}' not found in library, skipping") continue total_adduct_mass += adduct_mass * adduct_count total_adduct_charge += adduct_charge * adduct_count if adduct_count == 1: adduct_formula_parts.append(adduct_name) else: adduct_formula_parts.append(f'{adduct_count}{adduct_name}') inline_formula = adduct_entry.get('formula') try: if inline_formula: adduct_comp = composition_from_formula(inline_formula) total_multiplier = adduct_count else: base_match = re.match(r'^(\d+)?(.+)$', adduct_name) if base_match and base_match.group(1): inherent_count = int(base_match.group(1)) base_adduct = base_match.group(2) else: inherent_count = 1 base_adduct = adduct_name adduct_comp = composition_from_formula(base_adduct) total_multiplier = inherent_count * adduct_count for element, count in adduct_comp.items(): adduct_elements[element] = adduct_elements.get(element, 0) + (count * total_multiplier) except Exception as e: logger.warning(f"Could not parse adduct '{adduct_name}': {e}") logger.debug( f'Adduct: {adduct_count}×{adduct_name}: mass={adduct_mass * adduct_count:.4f} Da, charge={adduct_charge * adduct_count:+d}' ) adduct_string = '+'.join(adduct_formula_parts) if adduct_formula_parts else '' return total_adduct_mass, total_adduct_charge, adduct_string, adduct_elements def convert_numpy_types(obj: Any) -> Any: """ Recursively convert NumPy types to native Python types for JSON serialization. Handles nested dictionaries, lists, and arrays. Also converts Infinity and NaN to None for valid JSON. """ if isinstance(obj, np.integer): return int(obj) elif isinstance(obj, np.floating): val = float(obj) # Convert Infinity and NaN to None (null in JSON) if np.isnan(val) or np.isinf(val): return None return val elif isinstance(obj, (float, int)): # Handle native Python float/int that might be Infinity or NaN if isinstance(obj, float) and (np.isnan(obj) or np.isinf(obj)): return None return obj elif isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, dict): return {key: convert_numpy_types(value) for key, value in obj.items()} elif isinstance(obj, list): return [convert_numpy_types(item) for item in obj] elif isinstance(obj, tuple): return tuple(convert_numpy_types(item) for item in obj) else: return obj def to_subscript(n: int | str) -> str: """Convert number to subscript format. E.g., 1 → ₁, 28 → ₂₈""" subscript_map = str.maketrans('0123456789', '₀₁₂₃₄₅₆₇₈₉') return str(n).translate(subscript_map) app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB max file size # SECRET_KEY: Use environment variable in production # In development, use a random key; in production, require explicit setting _secret_key = os.environ.get('SECRET_KEY') if _secret_key: app.config['SECRET_KEY'] = _secret_key else: # Development only - generate random key (will change on restart) import secrets app.config['SECRET_KEY'] = secrets.token_hex(32) if os.environ.get('FLASK_ENV') == 'production': logger.warning('SECRET_KEY not set in production! Set SECRET_KEY environment variable.') # Session cookie security settings app.config['SESSION_COOKIE_SECURE'] = os.environ.get('FLASK_ENV') == 'production' # HTTPS only in production app.config['SESSION_COOKIE_HTTPONLY'] = True # Prevent JavaScript access app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' # CSRF protection # Simple rate limiting for analysis endpoints (prevents abuse) # Stores: {ip_address: [timestamp1, timestamp2, ...]} _rate_limit_requests: dict[str, list[float]] = {} _RATE_LIMIT_WINDOW = 60 # seconds _RATE_LIMIT_MAX_REQUESTS = 30 # max analysis requests per window per IP def check_rate_limit(ip_address: str | None) -> bool: """Check if IP has exceeded rate limit. Returns True if allowed, False if blocked.""" import time now = time.time() if ip_address is None: return True # Clean old entries if ip_address in _rate_limit_requests: _rate_limit_requests[ip_address] = [t for t in _rate_limit_requests[ip_address] if now - t < _RATE_LIMIT_WINDOW] else: _rate_limit_requests[ip_address] = [] # Check limit if len(_rate_limit_requests[ip_address]) >= _RATE_LIMIT_MAX_REQUESTS: return False # Record this request _rate_limit_requests[ip_address].append(now) return True # Input validation functions import html import re def validate_dna_sequence(sequence: str | None) -> tuple[bool, str | None]: """Validate DNA sequence contains only valid bases (ATCG)""" if not sequence: return False, 'Sequence cannot be empty' if not re.match(r'^[ATCG]+$', sequence.upper()): return False, 'Sequence must contain only A, T, C, G bases' if len(sequence) > 1000: return False, 'Sequence too long (max 1000 bases)' return True, None def validate_chemical_formula(formula: str | None) -> tuple[bool, str | None]: """Validate chemical formula format""" if not formula: return False, 'Formula cannot be empty' # Allow element symbols followed by optional numbers, with subscripts if not re.match(r'^[A-Za-z0-9₀₁₂₃₄₅₆₇₈₉]+$', formula): return False, 'Invalid formula format' if len(formula) > 500: return False, 'Formula too long' return True, None def validate_element_symbol(symbol: str | None) -> tuple[bool, str | None]: """Validate element symbol (e.g., Ag, Na, K)""" if not symbol: return False, 'Element symbol cannot be empty' if not re.match(r'^[A-Z][a-z]?$', symbol): return False, 'Invalid element symbol format' return True, None def validate_numeric_param(value: Any, min_val: float, max_val: float, name: str) -> tuple[bool, str | None]: """Validate numeric parameter is within range""" try: num = float(value) if num < min_val or num > max_val: return False, f'{name} must be between {min_val} and {max_val}' return True, None except (ValueError, TypeError): return False, f'{name} must be a number' def sanitize_string(s: Any, max_length: int = 100) -> str: """Sanitize string input - escape HTML and limit length""" if s is None: return '' s = str(s)[:max_length] return html.escape(s) # CSRF protection for JSON API endpoints def check_same_origin(f: F) -> F: """Decorator to verify request comes from same origin (CSRF protection for APIs)""" from functools import wraps @wraps(f) def decorated_function(*args: Any, **kwargs: Any) -> Any: # For JSON APIs, verify the request has proper content type # Browsers won't send application/json cross-origin without CORS preflight content_type = request.content_type or '' if request.method == 'POST' and 'application/json' not in content_type: return jsonify({'error': 'Invalid content type'}), 400 return f(*args, **kwargs) return decorated_function # type: ignore[return-value] # Global cache for isotope patterns - major speed optimization # Key: (formula, charge, resolution) -> Value: isotope pattern dict _isotope_pattern_cache: dict[tuple[str, int, int], dict[str, Any]] = {} _ISOTOPE_CACHE_MAX_SIZE = 500 # Limit cache size to prevent memory issues analyzer = DNASilverAnalyzer() @app.route('/add_manual_composition_by_formula', methods=['POST']) def add_manual_composition_by_formula() -> FlaskResponse: """Add a user-specified composition using ion formula directly""" try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] peak_mz = float(data.get('peak_mz')) charge = int(data.get('charge')) intensity = float(data.get('intensity')) formula = data.get('formula', '').strip() dna_sequence = data.get('dna_sequence', '') resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') custom_xna = data.get('custom_xna', None) # Get XNA settings logger.debug(f'Manual composition by formula - Formula: {formula}, XNA mode: {custom_xna is not None}') # For XNA mode, sequence is not required if not custom_xna and not dna_sequence: return jsonify({'error': 'DNA sequence is required'}), 400 if not spectrum_data: return jsonify({'error': 'No spectrum data available'}), 400 if not formula: return jsonify({'error': 'Ion formula is required'}), 400 # User enters the ION formula directly (e.g., C194H245N86O112P18Ag12 or C194H237N70O120P18Ag16NH4) # This is what's actually observed in MS after ionization - use it as-is! import re ion_formula = formula.strip() # Simple element parsing just to get nAg and nP for display element_pattern = r'([A-Z][a-z]?)(\d*)' elements = re.findall(element_pattern, ion_formula) num_silver = 0 num_phosphorus = 0 for element, count in elements: if element == 'Ag': num_silver = int(count) if count else 1 elif element == 'P': num_phosphorus = int(count) if count else 1 # MANUAL FORMULA ENTRY: Don't calculate N0/Qcl or parse adducts # User provides complete ion formula - we just use it directly for isotope pattern num_strands = None qcl = None n0 = None # For XNA mode, calculate corrected mass for pattern shifting user_neutral_mass = None if ( custom_xna and custom_xna.get('formula') and custom_xna.get('molecular_weight') is not None and num_phosphorus > 0 ): # Estimate num_strands from phosphorus count # Each nucleotide has 1 P, so P count = sequence_length * num_strands # For a typical 12-base sequence: P=12 means 1 strand, P=24 means 2 strands from pythoms.molecule import composition_from_formula xna_composition = composition_from_formula(custom_xna['formula']) p_per_strand = xna_composition.get('P', 0) if p_per_strand > 0: # Estimate number of strands from phosphorus count estimated_strands_raw = num_phosphorus / p_per_strand estimated_strands = round(estimated_strands_raw) # Warn if not a clean integer (formula might not match XNA composition) if abs(estimated_strands_raw - estimated_strands) > 0.1: logger.warning( f"P count ({num_phosphorus}) doesn't divide evenly by P per strand ({p_per_strand}). Calculated: {estimated_strands_raw:.2f} strands → Rounded to: {estimated_strands}. The entered formula may not match the XNA composition!" ) # Calculate corrected mass using XNA molecular weight mXNA_one = custom_xna['molecular_weight'] mDNA_total = mXNA_one * estimated_strands mAg_total = analyzer.mAg * num_silver # User neutral mass is XNA + Ag (before ionization) # This is compared with the theoretical mass from the formula to calculate shift user_neutral_mass = mDNA_total + mAg_total logger.info( f'XNA mode - estimated {estimated_strands} strands from P count {num_phosphorus}, XNA formula: {custom_xna["formula"]} (P={p_per_strand}), Total neutral mass: {user_neutral_mass:.2f} Da' ) else: logger.warning('XNA formula has no P atoms, cannot estimate num_strands') logger.debug(f'Manual formula entry - using formula as-is: {ion_formula}') # Calculate expected m/z using PythoMS formula parser # This handles any formula including adducts from pythoms.molecule import Molecule try: mol = Molecule(ion_formula) mass = mol.mass expected_mz = mass / charge mass_error_ppm = abs((expected_mz - peak_mz) / peak_mz * 1e6) except Exception: # Fallback if formula parsing fails expected_mz = peak_mz mass_error_ppm = 0.0 # Generate isotope pattern using the ion formula (as-is from user) mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) theo_pattern = analyzer.generate_isotope_pattern(ion_formula, charge, resolution) if 'error' not in theo_pattern: # Extract experimental data around the peak window = 3.0 mask = (mz_values >= peak_mz - window) & (mz_values <= peak_mz + window) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] theo_mz = theo_pattern['gaussian_mz'] theo_intensity = theo_pattern['gaussian_intensity'] # Calculate theoretical X0 from smooth Gaussian pattern (same method as exp_x0) theo_mz_gaussian = np.array(theo_pattern['gaussian_mz']) theo_int_gaussian = np.array(theo_pattern['gaussian_intensity']) if len(theo_mz_gaussian) > 0 and np.sum(theo_int_gaussian) > 0: theo_fit_result = analyzer.gaussian_fit_centroid(theo_mz_gaussian, theo_int_gaussian) if theo_fit_result and theo_fit_result[0] is not None: theo_x0 = theo_fit_result[0] theo_sigma = theo_fit_result[2] if len(theo_fit_result) > 2 else None else: # Fallback to weighted average if Gaussian fit fails theo_x0 = np.sum(theo_mz_gaussian * theo_int_gaussian) / np.sum(theo_int_gaussian) theo_sigma = None else: theo_x0, theo_sigma = None, None # Check if exp_x0 was provided from manual fit (frontend sends it) provided_exp_x0 = data.get('exp_x0') if provided_exp_x0 is not None: # Use the provided exp_x0 from manual fit instead of recalculating exp_x0 = float(provided_exp_x0) exp_sigma = None # We don't recalculate sigma here logger.debug( f'[add_manual_composition_by_formula] Using provided exp_x0 = {exp_x0:.4f} from manual fit' ) # Calculate X0 error using the provided (manual fit) exp_x0 if theo_x0 is not None: x0_error = abs(theo_x0 - exp_x0) else: x0_error = 999.0 else: # Calculate experimental X0 from Gaussian envelope (automatic fit) if len(exp_mz_window) > 0: # Generate Gaussian envelope for experimental data exp_mz_gaussian, exp_int_gaussian = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) if exp_mz_gaussian is not None and exp_int_gaussian is not None: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_gaussian, exp_int_gaussian) else: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_window, exp_int_window) # Calculate X0 error as: |theo_x0 - exp_x0| if theo_x0 is not None and exp_x0 is not None: x0_error = abs(theo_x0 - exp_x0) else: x0_error = 999.0 else: x0_error = 999.0 exp_x0, exp_sigma = None, None else: x0_error = 999.0 theo_mz = [] theo_intensity = [] theo_x0, theo_sigma = None, None exp_x0, exp_sigma = None, None # For manual formula entry: use the ion formula as-is (user-provided) # Don't try to reconstruct neutral formula since we're not calculating N0/Qcl display_formula = ion_formula logger.debug(f'Using ion formula for display: {display_formula}') # Build composition object - simplified for manual entry # Since N0/Qcl are always null, just show basic info full_notation = f'{display_formula} (z={charge})' # Default type based on silver content comp_type = 'nanocluster' if num_silver >= 2 else 'dna_ag_ion' composition = { 'type': comp_type, 'num_strands': None, # Not calculated for manual formulas 'num_silver': num_silver, 'qcl': None, # Not calculated for manual formulas 'n0': None, # Not calculated for manual formulas 'z': charge, 'formula': display_formula, # Display ion formula as-is 'ion_formula': display_formula, # Same - user provided ion formula 'neutral_formula': None, # Not calculated for manual formulas 'adduct': '', # No separate adduct notation 'full_notation': full_notation, 'expected_mz': expected_mz, 'mass_error_ppm': mass_error_ppm, 'x0_error': x0_error, 'abs_x0_error': abs(x0_error) if x0_error is not None else 999.0, 'theo_mz': theo_mz, 'theo_intensity': theo_intensity, 'theo_x0': float(theo_x0) if theo_x0 is not None else None, 'theo_sigma': float(theo_sigma) if theo_sigma is not None else None, 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'nH': 0, 'nC': 0, 'nN': 0, 'nO': 0, 'nP': 0, # Not parsed individually 'manual': True, # Flag to indicate this was manually added (skip X₀ threshold) } return jsonify(convert_numpy_types({'composition': composition})) except Exception as e: logger.error(f'ANALYSIS_CRASH in add_manual_composition_by_formula: {type(e).__name__}: {str(e)}') import traceback traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/add_manual_composition', methods=['POST']) def add_manual_composition() -> FlaskResponse: """Add a user-specified composition to a peak analysis""" try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] peak_mz = float(data.get('peak_mz')) charge = int(data.get('charge')) intensity = float(data.get('intensity')) num_strands = int(data.get('num_strands')) num_silver = int(data.get('num_silver')) qcl = int(data.get('qcl')) dna_sequence = data.get('dna_sequence', '') resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') custom_xna = data.get('custom_xna', None) # Get XNA settings # Get adduct information (array of {name, count}) # Example: [{"name": "NH4", "count": 1}, {"name": "Na", "count": 2}] adducts_input = data.get('adducts', []) if adducts_input: total_adduct_mass, total_adduct_charge, adduct_string, adduct_elements = parse_adduct_items( adducts_input, analyzer.adducts ) logger.info( f'Total adduct: {adduct_string} (mass={total_adduct_mass:.4f} Da, charge={total_adduct_charge:+d})' ) else: total_adduct_mass = 0.0 total_adduct_charge = 0 adduct_string = '' adduct_elements = {} # For XNA mode, sequence is not required if not custom_xna and not dna_sequence: return jsonify({'error': 'DNA sequence is required'}), 400 if not spectrum_data: return jsonify({'error': 'No spectrum data available'}), 400 # Calculate N0 # Formula: N₀ + Qcl = nAg (always, regardless of adducts) # Therefore: N₀ = nAg - Qcl n0 = num_silver - qcl if n0 < 0: return jsonify({'error': 'Invalid composition: N0 must be >= 0 (N0 = nAg - Qcl)'}), 400 # Get strand_type for complex mode (strand1, strand2, or complex) strand_type = data.get('strand_type', None) # Calculate composition (DNA or XNA) user_neutral_mass = None if custom_xna and custom_xna.get('formula'): # XNA mode: Use custom formula and molecular weight from pythoms.molecule import composition_from_formula # For complex mode, select the appropriate formula based on strand_type is_complex_mode = custom_xna.get('is_complex', False) if is_complex_mode and strand_type in ['strand1', 'strand2']: # Use individual strand formula strand1_formula = custom_xna.get('strand1_formula', '') or custom_xna.get('formula', '') strand2_formula = ( custom_xna.get('strand2_formula', '') or strand1_formula ) # Fallback to strand1 if same strands if strand_type == 'strand1': xna_formula = strand1_formula logger.info(f'Complex XNA mode - using strand1 formula: {xna_formula}') else: # strand_type == 'strand2' xna_formula = strand2_formula logger.info(f'Complex XNA mode - using strand2 formula: {xna_formula}') else: # Use combined formula (default for complex or regular XNA mode) xna_formula = custom_xna['formula'] if is_complex_mode: logger.info(f'Complex mode - using combined complex formula: {xna_formula}') xna_composition = composition_from_formula(xna_formula) nH = xna_composition.get('H', 0) * num_strands nC = xna_composition.get('C', 0) * num_strands nN = xna_composition.get('N', 0) * num_strands nO = xna_composition.get('O', 0) * num_strands nP = xna_composition.get('P', 0) * num_strands # Calculate masses from elements (for isotope pattern shape) mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mAg_total = analyzer.mAg * num_silver # Use user-provided molecular weight for mass calculation mXNA_one = custom_xna.get('molecular_weight') if mXNA_one is None: mXNA_one = analyzer.calculate_mass_from_formula(xna_formula) mDNA_total = mXNA_one * num_strands # Calculate user_neutral_mass for pattern shifting # This is the NEUTRAL mass (before ionization) used to position the isotope pattern # Include adduct mass in the neutral mass calculation user_neutral_mass = mDNA_total + mAg_total + total_adduct_mass logger.info( f'XNA mode - using custom molecular weight: {xna_formula}, Total neutral mass: {user_neutral_mass:.2f} Da' ) # Calculate expected m/z using corrected XNA mass WITH adduct neutral_mass = mDNA_total + mAg_total + total_adduct_mass mass = neutral_mass - (qcl + charge + total_adduct_charge) * analyzer.m_p elif custom_xna and custom_xna.get('is_complex') and not custom_xna.get('formula'): # DNA-only Complex mode: use appropriate sequence based on strand_type seq1 = custom_xna.get('dna_sequence1', dna_sequence) seq2 = custom_xna.get('dna_sequence2', seq1) if strand_type == 'strand2': # Single strand 2 only - works like normal DNA mode seq_to_use = seq2 logger.info(f'DNA Complex mode - strand2 only: {seq_to_use[:20]}...') elif strand_type == 'strand1': # Single strand 1 only - works like normal DNA mode seq_to_use = seq1 logger.info(f'DNA Complex mode - strand1 only: {seq_to_use[:20]}...') else: # strand_type == 'complex': Full complex # User enters total strands (2 strands = 1 complex) seq_to_use = seq1 logger.info(f'DNA Complex mode - complex (seq1 × {num_strands}): {seq_to_use[:20]}...') nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(seq_to_use, num_strands) # Calculate masses mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mAg_total = analyzer.mAg * num_silver mDNA_total = mP_total + mH_total + mC_total + mN_total + mO_total # Calculate expected m/z using the standard formula WITH adduct neutral_mass = mDNA_total + mAg_total + total_adduct_mass mass = neutral_mass - (qcl + charge + total_adduct_charge) * analyzer.m_p else: # DNA mode: Calculate from sequence nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(dna_sequence, num_strands) # Calculate masses mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mAg_total = analyzer.mAg * num_silver mDNA_total = mP_total + mH_total + mC_total + mN_total + mO_total # Calculate expected m/z using the standard formula WITH adduct # protons_removed = Qcl + z + adduct_charge neutral_mass = mDNA_total + mAg_total + total_adduct_mass mass = neutral_mass - (qcl + charge + total_adduct_charge) * analyzer.m_p expected_mz = mass / charge mass_error_ppm = abs((expected_mz - peak_mz) / peak_mz * 1e6) # Build formulas is_dna_only = num_silver == 0 # Calculate protons removed (accounting for adduct charge) # protons_removed = Qcl + z + adduct_charge protons_removed = (qcl + charge + total_adduct_charge) if not is_dna_only else (charge + total_adduct_charge) # Add adduct elements to base composition for ion formula nH_total_with_adduct = nH + adduct_elements.get('H', 0) nC_total_with_adduct = nC + adduct_elements.get('C', 0) nN_total_with_adduct = nN + adduct_elements.get('N', 0) nO_total_with_adduct = nO + adduct_elements.get('O', 0) nP_total_with_adduct = nP + adduct_elements.get('P', 0) nCl_with_adduct = adduct_elements.get('Cl', 0) nNa_with_adduct = adduct_elements.get('Na', 0) nK_with_adduct = adduct_elements.get('K', 0) if custom_xna: # XNA formula display xna_name = custom_xna['name'] if is_dna_only: neutral_formula = f'({xna_name}){to_subscript(num_strands)}' else: neutral_formula = f'({xna_name}){to_subscript(num_strands)}Ag{to_subscript(num_silver)}' # Add adduct to display formula if adduct_string: neutral_formula = f'{neutral_formula}+{adduct_string}' # Ion formula for isotope pattern (element-based + adducts) nH_ion = nH_total_with_adduct - protons_removed ion_formula = ( f'C{nC_total_with_adduct}H{nH_ion}N{nN_total_with_adduct}O{nO_total_with_adduct}P{nP_total_with_adduct}' ) if num_silver > 0: ion_formula += f'Ag{num_silver}' if nCl_with_adduct > 0: ion_formula += f'Cl{nCl_with_adduct}' if nCl_with_adduct > 1 else 'Cl' if nNa_with_adduct > 0: ion_formula += f'Na{nNa_with_adduct}' if nNa_with_adduct > 1 else 'Na' if nK_with_adduct > 0: ion_formula += f'K{nK_with_adduct}' if nK_with_adduct > 1 else 'K' else: # DNA formula display if is_dna_only: neutral_formula = f'C{nC}H{nH}N{nN}O{nO}P{nP}' nH_ion = nH_total_with_adduct - protons_removed ion_formula = f'C{nC_total_with_adduct}H{nH_ion}N{nN_total_with_adduct}O{nO_total_with_adduct}P{nP_total_with_adduct}' else: neutral_formula = f'C{nC}H{nH}N{nN}O{nO}P{nP}Ag{num_silver}' nH_ion = nH_total_with_adduct - protons_removed ion_formula = f'C{nC_total_with_adduct}H{nH_ion}N{nN_total_with_adduct}O{nO_total_with_adduct}P{nP_total_with_adduct}Ag{num_silver}' # Add adduct to display formula if adduct_string: neutral_formula = f'{neutral_formula}+{adduct_string}' # Add adduct elements to ion formula if nCl_with_adduct > 0: ion_formula += f'Cl{nCl_with_adduct}' if nCl_with_adduct > 1 else 'Cl' if nNa_with_adduct > 0: ion_formula += f'Na{nNa_with_adduct}' if nNa_with_adduct > 1 else 'Na' if nK_with_adduct > 0: ion_formula += f'K{nK_with_adduct}' if nK_with_adduct > 1 else 'K' # Generate isotope pattern and calculate X₀ error mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) theo_pattern = analyzer.generate_isotope_pattern(ion_formula, charge, resolution) if 'error' not in theo_pattern: theo_mz = np.array(theo_pattern['gaussian_mz']) theo_intensity = np.array(theo_pattern['gaussian_intensity']) # Theoretical X₀ from Gaussian centroid fit theo_x0, theo_sigma = None, None if len(theo_mz) > 0 and np.sum(theo_intensity) > 0: theo_fit_result = analyzer.gaussian_fit_centroid(theo_mz, theo_intensity) if theo_fit_result and theo_fit_result[0] is not None: theo_x0 = theo_fit_result[0] theo_sigma = theo_fit_result[2] if len(theo_fit_result) > 2 else None else: theo_x0 = np.sum(theo_mz * theo_intensity) / np.sum(theo_intensity) # Experimental Gaussian envelope window = 3.0 mask = (mz_values >= peak_mz - window) & (mz_values <= peak_mz + window) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] exp_mz_gaussian, exp_int_gaussian = None, None if len(exp_mz_window) > 0: exp_mz_gaussian, exp_int_gaussian = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) # Experimental X₀ provided_exp_x0 = data.get('exp_x0') if provided_exp_x0 is not None: exp_x0 = float(provided_exp_x0) exp_sigma = None elif exp_mz_gaussian is not None and exp_int_gaussian is not None: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_gaussian, exp_int_gaussian) elif len(exp_mz_window) > 0: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_window, exp_int_window) else: exp_x0, exp_sigma = None, None # X₀ error if theo_x0 is not None and exp_x0 is not None: x0_error = abs(theo_x0 - exp_x0) else: x0_error = 999.0 else: theo_mz = [] theo_intensity = [] theo_x0, theo_sigma = None, None exp_x0, exp_sigma = None, None x0_error = 999.0 # Build composition object # For display: displayed_qcl = qcl + total_adduct_charge displayed_qcl = qcl + total_adduct_charge if is_dna_only: comp_type = 'XNA Only' if custom_xna else 'DNA Only' full_notation = f'{neutral_formula} (z={charge})' else: comp_type = 'nanocluster' full_notation = f'{neutral_formula}-{qcl + charge}H (z={charge}, Qcl={displayed_qcl}, N0={n0})' composition = { 'type': comp_type, 'num_strands': num_strands, 'num_silver': num_silver, 'qcl': qcl, # Internal Qcl (N₀ + Qcl = nAg always) 'displayed_qcl': displayed_qcl, # For display: qcl + adduct_charge 'n0': n0, 'z': charge, 'formula': neutral_formula, 'ion_formula': ion_formula, 'neutral_formula': neutral_formula, 'adduct': adduct_string, # Include adduct information 'adduct_charge': total_adduct_charge, # Include for N0+Qcl relation display 'full_notation': full_notation, 'expected_mz': expected_mz, 'mass_error_ppm': mass_error_ppm, 'x0_error': x0_error, 'abs_x0_error': abs(x0_error) if x0_error is not None else 999.0, 'theo_mz': theo_mz, 'theo_intensity': theo_intensity, 'theo_x0': float(theo_x0) if theo_x0 is not None else None, 'theo_sigma': float(theo_sigma) if theo_sigma is not None else None, 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'nH': nH, 'nC': nC, 'nN': nN, 'nO': nO, 'nP': nP, 'manual': True, # Flag to indicate this was manually added (skip X₀ threshold) } return jsonify(convert_numpy_types({'composition': composition})) except Exception as e: logger.error(f'ANALYSIS_CRASH in add_manual_composition: {type(e).__name__}: {str(e)}') import traceback traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/add_manual_composition_search', methods=['POST']) def add_manual_composition_search() -> FlaskResponse: """Search for best N₀/Qcl composition with specified adduct""" try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] peak_mz = float(data.get('peak_mz')) charge = int(data.get('charge')) intensity = float(data.get('intensity')) num_strands = int(data.get('num_strands')) num_silver = int(data.get('num_silver')) adducts_input = data.get('adducts', []) # Array of {name, count} dna_sequence = data.get('dna_sequence', '') resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') custom_xna = data.get('custom_xna', None) strand_type = data.get('strand_type', None) # 'strand1', 'strand2', 'complex', or None logger.info( f'SEARCH MODE: Finding best N₀ for specified adducts - nf={num_strands}, nAg={num_silver}, Peak: m/z={peak_mz:.4f}, z={charge}' ) # For XNA mode, sequence is not required if not custom_xna and not dna_sequence: return jsonify({'error': 'DNA sequence is required'}), 400 if not spectrum_data: return jsonify({'error': 'No spectrum data available'}), 400 if adducts_input: total_adduct_mass, total_adduct_charge, adduct_string, adduct_elements = parse_adduct_items( adducts_input, analyzer.adducts ) if adduct_string: logger.info( f'Total adducts: {adduct_string} (mass={total_adduct_mass:.4f} Da, charge={total_adduct_charge:+d})' ) else: total_adduct_mass = 0.0 total_adduct_charge = 0 adduct_string = '' adduct_elements = {} # Calculate base DNA/XNA composition if custom_xna and custom_xna.get('formula'): # XNA mode from pythoms.molecule import composition_from_formula # For complex mode, select the appropriate formula based on strand_type is_complex_mode = custom_xna.get('is_complex', False) if is_complex_mode and strand_type in ['strand1', 'strand2']: # Use individual strand formula strand1_formula = custom_xna.get('strand1_formula', '') or custom_xna.get('formula', '') strand2_formula = ( custom_xna.get('strand2_formula', '') or strand1_formula ) # Fallback to strand1 if same strands if strand_type == 'strand1': xna_formula = strand1_formula logger.info(f'Complex XNA search mode - using strand1 formula: {xna_formula}') else: # strand_type == 'strand2' xna_formula = strand2_formula logger.info(f'Complex XNA search mode - using strand2 formula: {xna_formula}') else: # Use combined formula (default for complex or regular XNA mode) xna_formula = custom_xna['formula'] if is_complex_mode: logger.info(f'Complex search mode - using combined complex formula: {xna_formula}') xna_composition = composition_from_formula(xna_formula) nH = xna_composition.get('H', 0) * num_strands nC = xna_composition.get('C', 0) * num_strands nN = xna_composition.get('N', 0) * num_strands nO = xna_composition.get('O', 0) * num_strands nP = xna_composition.get('P', 0) * num_strands # Get XNA molecular weight mXNA_one = custom_xna.get('molecular_weight') if mXNA_one is None: mXNA_one = analyzer.calculate_mass_from_formula(xna_formula) mDNA_total = mXNA_one * num_strands elif custom_xna and custom_xna.get('is_complex') and not custom_xna.get('formula'): # DNA-only Complex mode: use appropriate sequence based on strand_type seq1 = custom_xna.get('dna_sequence1', dna_sequence) seq2 = custom_xna.get('dna_sequence2', seq1) if strand_type == 'strand2': # Single strand 2 only - works like normal DNA mode seq_to_use = seq2 logger.info(f'DNA Complex search mode - strand2 only: {seq_to_use[:20]}...') elif strand_type == 'strand1': # Single strand 1 only - works like normal DNA mode seq_to_use = seq1 logger.info(f'DNA Complex search mode - strand1 only: {seq_to_use[:20]}...') else: # strand_type == 'complex': Full complex # User enters total strands (2 strands = 1 complex) seq_to_use = seq1 logger.info(f'DNA Complex search mode - complex (seq1 × {num_strands}): {seq_to_use[:20]}...') nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(seq_to_use, num_strands) mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mDNA_total = mP_total + mH_total + mC_total + mN_total + mO_total else: # DNA mode nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(dna_sequence, num_strands) mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mDNA_total = mP_total + mH_total + mC_total + mN_total + mO_total mAg_total = analyzer.mAg * num_silver # Prepare spectrum data for pattern matching mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) # Get manual fit range if provided manual_fit_range = data.get('manual_fit_range') provided_exp_x0 = data.get('exp_x0') # Search all N₀ values (qcl from 0 to nAg) all_compositions = [] for qcl in range(num_silver + 1): # Formula: N₀ + Qcl = nAg (always, regardless of adducts) # Therefore: N₀ = nAg - Qcl n0 = num_silver - qcl # Calculate mass for this qcl # protons_removed = Qcl + z + adduct_charge protons_removed = qcl + charge + total_adduct_charge if custom_xna: user_neutral_mass = mDNA_total + mAg_total + total_adduct_mass # Ion mass = neutral mass - mass of removed protons mass_ion = user_neutral_mass - (protons_removed * analyzer.m_p) else: user_neutral_mass = None # Ion mass = DNA + Ag + adducts - mass of removed protons mass_ion = mDNA_total + mAg_total + total_adduct_mass - (protons_removed * analyzer.m_p) expected_mz = mass_ion / charge mass_error_ppm = abs((expected_mz - peak_mz) / peak_mz * 1e6) # Build formulas protons_removed = qcl + charge + total_adduct_charge # Add adduct elements to base counts (for H, C, N, O, P which are in the base formula) nH_total_with_adduct = nH + adduct_elements.get('H', 0) nC_total_with_adduct = nC + adduct_elements.get('C', 0) nN_total_with_adduct = nN + adduct_elements.get('N', 0) nO_total_with_adduct = nO + adduct_elements.get('O', 0) nP_total_with_adduct = nP + adduct_elements.get('P', 0) # Build formulas if custom_xna: xna_name = custom_xna['name'] neutral_formula = f'({xna_name}){to_subscript(num_strands)}Ag{to_subscript(num_silver)}' if adduct_string: neutral_formula = f'{neutral_formula}+{adduct_string}' nH_ion = nH_total_with_adduct - protons_removed ion_formula = f'C{nC_total_with_adduct}H{nH_ion}N{nN_total_with_adduct}O{nO_total_with_adduct}P{nP_total_with_adduct}Ag{num_silver}' else: neutral_formula = f'C{nC}H{nH}N{nN}O{nO}P{nP}Ag{num_silver}' if adduct_string: neutral_formula = f'{neutral_formula}+{adduct_string}' nH_ion = nH_total_with_adduct - protons_removed ion_formula = f'C{nC_total_with_adduct}H{nH_ion}N{nN_total_with_adduct}O{nO_total_with_adduct}P{nP_total_with_adduct}Ag{num_silver}' # Add ALL adduct elements to ion formula (handles any element: Cl, Na, K, Br, I, etc.) # Skip elements already in base formula (C, H, N, O, P, Ag) base_elements = {'C', 'H', 'N', 'O', 'P', 'Ag'} for element, count in adduct_elements.items(): if element not in base_elements and count > 0: ion_formula += f'{element}{count}' if count > 1 else element # Generate isotope pattern theo_pattern = analyzer.generate_isotope_pattern(ion_formula, charge, resolution) if 'error' not in theo_pattern: theo_mz = np.array(theo_pattern['gaussian_mz']) theo_intensity = np.array(theo_pattern['gaussian_intensity']) # Theoretical X₀ from Gaussian centroid fit theo_x0, theo_sigma = None, None if len(theo_mz) > 0 and np.sum(theo_intensity) > 0: theo_fit_result = analyzer.gaussian_fit_centroid(theo_mz, theo_intensity) if theo_fit_result and theo_fit_result[0] is not None: theo_x0 = theo_fit_result[0] theo_sigma = theo_fit_result[2] if len(theo_fit_result) > 2 else None else: theo_x0 = np.sum(theo_mz * theo_intensity) / np.sum(theo_intensity) # Experimental Gaussian envelope window = 3.0 mask = (mz_values >= peak_mz - window) & (mz_values <= peak_mz + window) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] exp_mz_gaussian, exp_int_gaussian = None, None if len(exp_mz_window) > 0: exp_mz_gaussian, exp_int_gaussian = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) # Experimental X₀ if provided_exp_x0 is not None: exp_x0 = float(provided_exp_x0) exp_sigma = None elif exp_mz_gaussian is not None and exp_int_gaussian is not None: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_gaussian, exp_int_gaussian) elif len(exp_mz_window) > 0: exp_x0, exp_sigma, _ = analyzer.gaussian_fit_centroid(exp_mz_window, exp_int_window) else: exp_x0, exp_sigma = None, None # X₀ error if theo_x0 is not None and exp_x0 is not None: x0_error = abs(theo_x0 - exp_x0) else: x0_error = 999.0 # Pattern similarity (stick-vs-apex comparison) pattern_score = 0.0 if len(exp_mz_window) > 0: theo_stick_mz = np.array(theo_pattern['mz']) theo_stick_int = np.array(theo_pattern['intensity']) pattern_score = analyzer.calculate_pattern_similarity( theo_stick_mz, theo_stick_int, exp_mz_window, exp_int_window ) else: theo_mz = [] theo_intensity = [] theo_x0, theo_sigma = None, None exp_x0, exp_sigma = None, None x0_error = 999.0 pattern_score = 0.0 # Build composition object # For display: displayed_qcl = qcl + total_adduct_charge displayed_qcl = qcl + total_adduct_charge full_notation = f'{neutral_formula}-{qcl + charge}H (z={charge}, Qcl={displayed_qcl}, N0={n0})' composition = { 'type': 'nanocluster', 'num_strands': num_strands, 'num_silver': num_silver, 'qcl': qcl, # Internal Qcl (N₀ + Qcl = nAg always) 'displayed_qcl': displayed_qcl, # For display: qcl + adduct_charge 'n0': n0, 'z': charge, 'formula': neutral_formula, 'ion_formula': ion_formula, 'neutral_formula': neutral_formula, 'adduct': adduct_string, 'adduct_charge': total_adduct_charge, # Include for N0+Qcl relation display 'full_notation': full_notation, 'expected_mz': expected_mz, 'mass_error_ppm': mass_error_ppm, 'x0_error': x0_error, 'abs_x0_error': abs(x0_error) if x0_error is not None else 999.0, 'pattern_score': pattern_score, 'theo_mz': theo_mz, 'theo_intensity': theo_intensity, 'theo_x0': float(theo_x0) if theo_x0 is not None else None, 'theo_sigma': float(theo_sigma) if theo_sigma is not None else None, 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'nH': nH, 'nC': nC, 'nN': nN, 'nO': nO, 'nP': nP, 'custom_xna': custom_xna, 'manual': True, # Flag to indicate this was manually added (skip X₀ threshold) } all_compositions.append(composition) logger.debug(f'N₀={n0} (Qcl={qcl}): X₀_error={x0_error:.4f} m/z, pattern_score={pattern_score:.3f}') # Sort ALL compositions by X₀ error (lowest first) - this is the primary metric all_compositions.sort(key=lambda c: c['abs_x0_error']) # Show all N₀ values searched (for debugging) logger.debug(f'All {len(all_compositions)} compositions sorted by X₀ error') # The true best is now the first one (lowest X₀ error) true_best = all_compositions[0] logger.info( f'Best composition (by X₀ error): N₀={true_best["n0"]}, Qcl={true_best["qcl"]}, X₀ error: {true_best["x0_error"]:.4f} m/z, Pattern score: {true_best["pattern_score"]:.3f}' ) # COMPLEX MODE: Only return N₀=0 composition (Qcl = nAg) # All complex strand types (strand1, strand2, complex, nd=X) have N₀=0 constraint is_complex = strand_type and (strand_type in ['strand1', 'strand2', 'complex'] or strand_type.startswith('nd=')) if is_complex: # For complex mode, N₀ = 0 always, so Qcl = nAg complex_comp = next((c for c in all_compositions if c['n0'] == 0), None) if complex_comp: logger.info( f'COMPLEX MODE: Returning only N₀=0 composition - nAg={complex_comp["num_silver"]}, Qcl={complex_comp["qcl"]}, X₀_err={complex_comp["x0_error"]:.4f}' ) return jsonify(convert_numpy_types({'compositions': [complex_comp]})) else: logger.info('COMPLEX MODE: No N₀=0 composition found') return jsonify({'compositions': []}) # NON-COMPLEX MODE: Return exactly 3 compositions (best Qcl, Qcl+1, Qcl-1) best_qcl = true_best['qcl'] # Build list with best composition first, then Qcl±1 result_compositions = [true_best] # Best composition (lowest X₀ error) # Find Qcl-1 composition qcl_minus_1 = next((c for c in all_compositions if c['qcl'] == best_qcl - 1), None) if qcl_minus_1: result_compositions.append(qcl_minus_1) # Find Qcl+1 composition qcl_plus_1 = next((c for c in all_compositions if c['qcl'] == best_qcl + 1), None) if qcl_plus_1: result_compositions.append(qcl_plus_1) # Sort by X₀ error (best first) to maintain consistent ranking result_compositions.sort(key=lambda c: c['abs_x0_error']) logger.info(f'Returning {len(result_compositions)} compositions (best Qcl={best_qcl} ± 1)') return jsonify(convert_numpy_types({'compositions': result_compositions})) except Exception as e: logger.error(f'ANALYSIS_CRASH in add_manual_composition_search: {type(e).__name__}: {str(e)}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/reanalyze_peak', methods=['POST']) @check_same_origin def reanalyze_peak() -> FlaskResponse: """Re-analyze a single peak with user-specified charge state""" # Rate limiting check if not check_rate_limit(request.remote_addr): return jsonify({'error': 'Rate limit exceeded. Please wait before making more requests.'}), 429 try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] peak_mz = float(data.get('peak_mz')) charge = int(data.get('charge')) intensity = float(data.get('intensity')) dna_sequence = data.get('dna_sequence', '') resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') custom_xna = data.get('custom_xna', None) # Get XNA settings if provided complex_mode = data.get('complex_mode', None) # Get complex mode settings if provided # Handle complex mode - pass complex settings to analyzer if complex_mode and complex_mode.get('enabled'): logger.debug('[reanalyze_peak] Complex mode enabled') # For complex mode, use the complex XNA settings if complex_mode.get('xna'): custom_xna = complex_mode['xna'] logger.debug(f'[reanalyze_peak] Using complex XNA: {custom_xna}') else: # DNA-only Complex mode: no XNA formula, but still flag as complex for N0=0 custom_xna = {'name': 'Complex', 'is_complex': True, 'same_strands': False} logger.debug(f'[reanalyze_peak] DNA-only Complex mode: {custom_xna}') if not spectrum_data: return jsonify({'error': 'No spectrum data available'}), 400 # Extract spectrum arrays mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) # Generate experimental Gaussian curve for X0 calculation # Extract experimental data around the peak window = 3.0 mask = (mz_values >= peak_mz - window) & (mz_values <= peak_mz + window) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] # Generate smooth Gaussian envelope (SAME AS AUTO ANALYSIS) exp_mz_gaussian, exp_int_gaussian = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) # Fit Gaussian using gaussian_fit_centroid (same method as custom search) if exp_mz_gaussian is not None and exp_int_gaussian is not None: fit_result = analyzer.gaussian_fit_centroid(exp_mz_gaussian, exp_int_gaussian) if fit_result and fit_result[0] is not None: exp_x0 = fit_result[0] exp_sigma = fit_result[1] else: exp_x0, exp_sigma = None, None else: exp_x0, exp_sigma = None, None if exp_x0 is None: logger.debug('[reanalyze_peak] Envelope generation failed, using weighted average') exp_x0, exp_sigma = analyzer.weighted_average_centroid(exp_mz_window, exp_int_window) exp_mz_gaussian = exp_mz_window exp_int_gaussian = exp_int_window # Use SMART composition finding with adduct search and X0 error threshold detection logger.info(f'[reanalyze_peak] Finding compositions with smart adduct search for exp_x0={exp_x0:.4f}') compositions = analyzer.analyze_peak_with_smart_adduct_search( peak_mz, charge, dna_sequence, exp_x0, resolution=resolution, mz_values=mz_values, intensity_values=intensity_values, custom_xna=custom_xna, ) # Refine with isotope matching has_other_strands = False all_compositions = [] has_odd_n0_warning = False if len(compositions) > 0: # Use underscores for returned Gaussian values - we keep the fitted ones calculated above ( refined_compositions, _, _, has_other_strands, all_compositions, has_odd_n0_warning, _, _, has_unrealistic_n0_warning, ) = analyzer.refine_compositions_with_isotope_matching( compositions=compositions, experimental_mz=mz_values, experimental_int=intensity_values, peak_mz=peak_mz, resolution=resolution, detected_centroid=exp_x0, # Use calculated X₀ ) logger.info(f'[reanalyze_peak] Refined {len(refined_compositions)} compositions') else: refined_compositions = [] logger.info('[reanalyze_peak] No compositions found') # Keep the fitted Gaussian curve for display (exp_mz_gaussian, exp_int_gaussian already set above) # Calculate peak symmetry symmetry_info = analyzer.calculate_peak_symmetry( mz_values=mz_values, intensity_values=intensity_values, center_mz=peak_mz, window=2.0 ) # Prepare Gaussian curve for display (already calculated above) exp_gaussian_mz: list[float] = [] exp_gaussian_intensity: list[float] = [] if exp_mz_gaussian is not None and exp_int_gaussian is not None: exp_gaussian_mz = exp_mz_gaussian.tolist() exp_gaussian_intensity = exp_int_gaussian.tolist() # Determine if this is complex mode is_complex = complex_mode and complex_mode.get('enabled', False) # Get max intensity around the peak for display max_int = float(np.max(exp_int_window)) if len(exp_int_window) > 0 else 0.0 # Convert all NumPy types to native Python types for JSON serialization result = { 'peak_mz': float(peak_mz), 'intensity': max_int, # Required for frontend display 'compositions': refined_compositions, 'charge': charge, 'charge_method': 'user_specified', 'charge_confidence': 1.0, # User specified = full confidence 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'symmetry': symmetry_info, 'has_other_strands': has_other_strands, 'all_compositions': all_compositions, 'exp_gaussian_mz': exp_gaussian_mz, 'exp_gaussian_intensity': exp_gaussian_intensity, 'is_complex': is_complex, } return jsonify(convert_numpy_types(result)) except Exception as e: logger.error(f'ANALYSIS_CRASH in reanalyze_peak: {type(e).__name__}: {str(e)}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/recalculate_peak_with_manual_fit', methods=['POST']) def recalculate_peak_with_manual_fit() -> FlaskResponse: """Recalculate peak analysis with user-specified m/z range for Gaussian fitting""" try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] peak_mz = float(data.get('peak_mz')) charge = int(data.get('charge')) intensity = float(data.get('intensity')) start_mz = float(data.get('start_mz')) end_mz = float(data.get('end_mz')) dna_sequence = data.get('dna_sequence', '') resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') custom_xna = data.get('custom_xna', None) # Get XNA settings complex_mode = data.get('complex_mode', None) # Get complex mode settings if provided # Handle complex mode - pass complex settings to analyzer if complex_mode and complex_mode.get('enabled'): logger.debug('[recalculate_peak_with_manual_fit] Complex mode enabled') # For complex mode, use the complex XNA settings if complex_mode.get('xna'): custom_xna = complex_mode['xna'] logger.debug(f'[recalculate_peak_with_manual_fit] Using complex XNA: {custom_xna}') else: # DNA-only Complex mode: no XNA formula, but still flag as complex for N0=0 custom_xna = {'name': 'Complex', 'is_complex': True, 'same_strands': False} logger.debug(f'[recalculate_peak_with_manual_fit] DNA-only Complex mode: {custom_xna}') if not spectrum_data: return jsonify({'error': 'No spectrum data available'}), 400 # Validate m/z range if start_mz >= end_mz: return jsonify({'error': 'Start m/z must be less than end m/z'}), 400 # Extract spectrum arrays mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) # Extract experimental data within the user-specified range mask = (mz_values >= start_mz) & (mz_values <= end_mz) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] if len(exp_mz_window) < 3: return jsonify({'error': 'Not enough data points in specified range. Please widen the range.'}), 400 logger.info( f'USER_ACTION: Manual Gaussian fit - Peak m/z: {peak_mz:.4f}, z={charge}, range: [{start_mz:.4f}, {end_mz:.4f}], {len(exp_mz_window)} points' ) # Generate smooth Gaussian envelope from user-specified range logger.debug('Generating smooth envelope from user-specified range') exp_mz_gaussian, exp_int_gaussian = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) # Fit Gaussian using gaussian_fit_centroid (same method as custom search) if exp_mz_gaussian is not None and exp_int_gaussian is not None: fit_result = analyzer.gaussian_fit_centroid(exp_mz_gaussian, exp_int_gaussian) if fit_result and fit_result[0] is not None: exp_x0 = fit_result[0] exp_sigma = fit_result[1] else: exp_x0, exp_sigma = None, None else: exp_x0, exp_sigma = None, None if exp_x0 is None: logger.debug('Envelope generation failed, using weighted average') exp_x0, exp_sigma = analyzer.weighted_average_centroid(exp_mz_window, exp_int_window) exp_mz_gaussian = exp_mz_window exp_int_gaussian = exp_int_window logger.debug(f'exp_x0={exp_x0:.4f} (fallback)') # Calculate symmetry (for info only) symmetry_info = analyzer.calculate_peak_symmetry( mz_values=mz_values, intensity_values=intensity_values, center_mz=peak_mz, window=2.0 ) symmetry_percent = symmetry_info.get('symmetry_score', 0.0) * 100 logger.debug(f'Peak symmetry: {symmetry_percent:.1f}%') # Use SAME composition finding as automatic analysis (with smart adduct search) logger.info(f'Finding compositions using exp_x0={exp_x0:.4f}') compositions = analyzer.analyze_peak_with_smart_adduct_search( peak_mz, charge, dna_sequence, exp_x0, resolution=resolution, mz_values=mz_values, intensity_values=intensity_values, custom_xna=custom_xna, # Pass XNA settings for pattern matching! ) # Refine with isotope matching - SAME AS AUTO ANALYSIS has_other_strands = False all_compositions = [] has_odd_n0_warning = False has_unrealistic_n0_warning = False if len(compositions) > 0: ( compositions, exp_x0_refined, exp_sigma_refined, has_other_strands, all_compositions, has_odd_n0_warning, _, _, has_unrealistic_n0_warning, ) = analyzer.refine_compositions_with_isotope_matching( compositions, mz_values, intensity_values, peak_mz, resolution=resolution, detected_centroid=exp_x0, # Use manual fit X₀ ) # Keep the manual fit X₀ (don't overwrite) # exp_x0 stays as is logger.info(f'Refined {len(compositions)} compositions using exp_x0={exp_x0:.4f}') # Sort compositions by combined score (pattern similarity + X0 error) # Higher score is better def combined_score_manual(comp): pattern_sim = comp.get('pattern_similarity', 0.0) x0_err = abs(comp.get('x0_error', 999.0)) if comp.get('x0_error') not in [None, 999.0] else 999.0 return pattern_sim - (x0_err * 0.1) compositions.sort(key=combined_score_manual, reverse=True) # reverse=True for descending (highest score first) logger.debug( f'Top {min(10, len(compositions))} compositions after manual fit recalculation (sorted by combined score)' ) # FILTER BY Qcl ± 1: Find best composition by SMALLEST X0 ERROR, then keep only Qcl ± 1 valid_comps = [c for c in compositions if c.get('x0_error', 999.0) != 999.0] if valid_comps: # Sort by SMALLEST X0 ERROR (for determining Qcl range) sorted_comps = sorted(valid_comps, key=lambda c: abs(c.get('x0_error', 999.0))) best_comp = sorted_comps[0] best_qcl = best_comp.get('qcl') if best_comp.get('type') == 'nanocluster' else None best_pattern_sim = best_comp.get('pattern_similarity', 0.0) best_x0_err = best_comp.get('x0_error', 999.0) logger.debug( f'Best X0 match for Qcl filtering: Type={best_comp.get("type")}, nAg={best_comp.get("num_silver")}, N0={best_comp.get("n0")}, Qcl={best_qcl}, X0_error={best_x0_err:.4f}' ) # Filter to keep only Qcl ± 1 of best composition (for nanoclusters) + all non-nanoclusters if best_qcl is not None: filtered_compositions = [ comp for comp in compositions if (comp.get('type') != 'nanocluster') # Keep all non-cluster or (comp.get('qcl') is not None and abs(comp['qcl'] - best_qcl) <= 1) # Qcl±1 of best ] nanocluster_count = len([c for c in filtered_compositions if c.get('type') == 'nanocluster']) non_cluster_count = len([c for c in filtered_compositions if c.get('type') != 'nanocluster']) logger.info( f'FILTERED: Kept {nanocluster_count} nanoclusters (Qcl±1 of {best_qcl}) + {non_cluster_count} non-cluster' ) compositions = filtered_compositions if len(filtered_compositions) > 0 else compositions else: logger.debug('Best composition is non-nanocluster - no Qcl filtering applied') else: logger.warning('No valid compositions with X0 error - returning all') # Re-sort filtered compositions by PATTERN SIMILARITY (highest first) # Now that we've filtered to correct Qcl range, rank by how well patterns match compositions.sort(key=combined_score_manual, reverse=True) # Highest score first # Take top matches after filtering and re-sorting refined_compositions = compositions[:10] if len(compositions) > 10 else compositions logger.info(f'Final compositions: {len(refined_compositions)} after Qcl±1 filter') # Convert to JSON-serializable format result = { 'compositions': refined_compositions, 'charge': charge, 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'symmetry': symmetry_info, 'manual_fit_range': [float(start_mz), float(end_mz)], 'exp_gaussian_mz': exp_mz_gaussian.tolist() if exp_mz_gaussian is not None else [], 'exp_gaussian_intensity': exp_int_gaussian.tolist() if exp_int_gaussian is not None else [], 'has_unrealistic_n0_warning': has_unrealistic_n0_warning, } return jsonify(convert_numpy_types(result)) except Exception as e: logger.error(f'ANALYSIS_CRASH in recalculate_peak_with_manual_fit: {type(e).__name__}: {str(e)}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/try_higher_strands', methods=['POST']) def try_higher_strands() -> FlaskResponse: """Try higher strand numbers (4-6) for a given peak""" try: data = request.get_json() peak_mz = float(data['peak_mz']) _custom_adducts_payload = data.get('custom_adducts', []) or [] dna_sequence = data.get('dna_sequence', '') charge = int(data.get('charge', 1)) # Get spectrum data from request mz_list = data.get('mz_values', []) intensity_list = data.get('intensity_values', []) if not dna_sequence: return jsonify({'error': 'DNA sequence required'}), 400 if not mz_list or not intensity_list: return jsonify({'error': 'No spectrum data provided'}), 400 analyzer = DNASilverAnalyzer() analyzer.custom_adducts = _custom_adducts_payload mz_values = np.array(mz_list) intensity_values = np.array(intensity_list) # Generate initial compositions for strands 1-3 to find max strand number logger.info(f'Finding current compositions for peak at m/z {peak_mz:.4f}') compositions = [] # First pass: generate compositions for strands 1-3 for num_strands in range(1, MAX_STRANDS + 1): for num_ag in range(2, MAX_SILVER + 1): nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(dna_sequence, num_strands) mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mAg_total = analyzer.mAg * num_ag z_values = [charge] if charge else [1, 2, 3, 4, 5, 6, 7, 8] for z_test in z_values: if z_test is None or z_test <= 0: continue for qcl in range(0, num_ag + 1): n0_valence = num_ag - qcl if n0_valence < 0: continue mass = ( mP_total + mH_total + mC_total + mN_total + mO_total + mAg_total - (qcl + z_test) * analyzer.m_p ) expected_mz = mass / z_test mass_error_ppm = abs((expected_mz - peak_mz) / peak_mz * 1e6) if mass_error_ppm < 200: compositions.append( { 'num_strands': num_strands, 'num_silver': num_ag, 'qcl': qcl, 'n0': n0_valence, 'mass_error_ppm': mass_error_ppm, } ) # Find the maximum strand number in current compositions max_strand = 0 for comp in compositions: if comp.get('num_strands', 0) > max_strand: max_strand = comp['num_strands'] # Search for next strand number (max + 1) next_strand = max_strand + 1 logger.info(f'Current max strand number: {max_strand}, searching for strand number: {next_strand}') # Now generate compositions for the next strand number # Start fresh with only the next strand number compositions = [] # Start from num_ag=0 to include DNA-only compositions for num_ag in range(0, MAX_SILVER + 1): num_strands = next_strand nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(dna_sequence, num_strands) mH_total = analyzer.m_p * nH mC_total = analyzer.mC * nC mN_total = analyzer.mN * nN mO_total = analyzer.mO * nO mP_total = analyzer.mP * nP mAg_total = analyzer.mAg * num_ag z_values = [charge] if charge else [1, 2, 3, 4, 5, 6, 7, 8] for z_test in z_values: if z_test is None or z_test <= 0: continue for qcl in range(0, num_ag + 1): n0_valence = num_ag - qcl if n0_valence < 0: continue mass = ( mP_total + mH_total + mC_total + mN_total + mO_total + mAg_total - (qcl + z_test) * analyzer.m_p ) expected_mz = mass / z_test mass_error_ppm = abs((expected_mz - peak_mz) / peak_mz * 1e6) if mass_error_ppm < 200: neutral_formula = f'C{nC}H{nH}N{nN}O{nO}P{nP}Ag{num_ag}' nH_ion = nH - (qcl + z_test) ion_formula = f'C{nC}H{nH_ion}N{nN}O{nO}P{nP}Ag{num_ag}' compositions.append( { 'type': 'nanocluster', 'num_strands': num_strands, 'num_silver': num_ag, 'qcl': qcl, 'n0': n0_valence, 'z': z_test, 'formula': neutral_formula, 'ion_formula': ion_formula, 'neutral_formula': neutral_formula, 'adduct': '', 'full_notation': f'{neutral_formula}-{qcl + z_test}H (z={z_test}, Qcl={qcl}, N0={n0_valence})', 'expected_mz': expected_mz, 'mass_error_ppm': mass_error_ppm, 'x0_error': 999.0, 'nH': nH, 'nC': nC, 'nN': nN, 'nO': nO, 'nP': nP, } ) logger.info(f'Found {len(compositions)} total compositions (strands 1-{next_strand})') # Refine with isotope matching ( refined_compositions, exp_x0, exp_sigma, has_other_strands, all_compositions, has_odd_n0_warning, _, _, has_unrealistic_n0_warning, ) = analyzer.refine_compositions_with_isotope_matching( compositions, mz_values, intensity_values, peak_mz, resolution=20000 ) # Calculate symmetry symmetry_info = analyzer.calculate_peak_symmetry( mz_values=mz_values, intensity_values=intensity_values, center_mz=peak_mz, window=2.0 ) # Count compositions by strand number strand_counts: dict[int, int] = {} for comp in refined_compositions: num_strands = comp.get('num_strands', 0) strand_counts[num_strands] = strand_counts.get(num_strands, 0) + 1 return jsonify( { 'compositions': refined_compositions, 'all_compositions': all_compositions, 'charge': charge, 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'symmetry': symmetry_info, 'has_other_strands': has_other_strands, 'strand_range': f'1-{next_strand}', 'strand_counts': strand_counts, 'next_strand': next_strand, 'has_unrealistic_n0_warning': has_unrealistic_n0_warning, } ) except Exception as e: logger.error(f'ANALYSIS_CRASH in try_higher_strands: {type(e).__name__}: {str(e)}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/') def index() -> str: """Render main page""" return render_template('index.html') @app.route('/calculate_dna_mass', methods=['POST']) def calculate_dna_mass() -> FlaskResponse: """Calculate the mass of single-stranded DNA from sequence""" try: data = request.json if not data: return jsonify({'error': 'No data provided'}), 400 dna_sequence = data.get('dna_sequence', '').upper().strip() if not dna_sequence: return jsonify({'error': 'No DNA sequence provided'}), 400 # Validate sequence (only ATCG allowed) valid_bases = set('ATCG') if not all(base in valid_bases for base in dna_sequence): return jsonify({'error': 'Invalid DNA sequence. Only A, T, C, G allowed.'}), 400 # Count bases base_count = { 'A': dna_sequence.count('A'), 'T': dna_sequence.count('T'), 'C': dna_sequence.count('C'), 'G': dna_sequence.count('G'), } # Calculate composition for single strand (nf=1) nH, nC, nN, nO, nP = analyzer.calculate_dna_composition(dna_sequence, strands=1) # Calculate total mass mass = analyzer.m_p * nH + analyzer.mC * nC + analyzer.mN * nN + analyzer.mO * nO + analyzer.mP * nP return jsonify( { 'mass': float(mass), 'length': len(dna_sequence), 'composition': {'H': nH, 'C': nC, 'N': nN, 'O': nO, 'P': nP}, 'base_count': base_count, 'formula': f'C{nC}H{nH}N{nN}O{nO}P{nP}', } ) except Exception as e: logger.error(f'ANALYSIS_CRASH in calculate_dna_mass: {type(e).__name__}: {str(e)}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/upload', methods=['POST']) def upload_spectrum() -> FlaskResponse: """Handle spectrum file upload with automatic peak detection and charge assignment""" # Rate limiting check if not check_rate_limit(request.remote_addr): return jsonify({'error': 'Rate limit exceeded. Please wait before uploading.'}), 429 try: # Clear caches when new spectrum is uploaded global _peak_analysis_cache, _isotope_pattern_cache peak_cache_size = len(_peak_analysis_cache) isotope_cache_size = len(_isotope_pattern_cache) _peak_analysis_cache.clear() _isotope_pattern_cache.clear() if peak_cache_size > 0 or isotope_cache_size > 0: logger.debug(f'Cleared caches (peak: {peak_cache_size}, isotope: {isotope_cache_size})') # Check if file was uploaded if 'file' not in request.files: return jsonify({'error': 'No file uploaded'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No file selected'}), 400 # Read file content content = file.read().decode('utf-8') # Parse spectrum mz_values, intensity_values = analyzer.parse_txt_spectrum(content) if len(mz_values) == 0: return jsonify({'error': 'No valid data found in file'}), 400 # Auto-detect resolution from spectrum estimated_resolution = analyzer.estimate_resolution(mz_values, intensity_values) logger.info(f'Auto-detected resolution: {estimated_resolution}') # AUTO-DETECT ALL PEAKS AND ASSIGN CHARGES (Senko et al. 1995 method) logger.info('Detecting peak regions (isotope envelopes) and assigning charge states...') detected_peaks = detect_all_peaks_with_charge( mz_values, intensity_values, prominence=0.01, charge_range=(1, 10), method='combination', merge_gap=1.5 ) logger.info(f'Found {len(detected_peaks)} isotope envelopes with charge states') # Cross-check each peak's charge with direct spacing measurement for peak in detected_peaks: if peak['charge'] is not None: spacing_check = analyzer.detect_charge_state(mz_values, intensity_values, peak['mz'], window=3.0) if ( spacing_check['charge'] is not None and spacing_check['num_peaks'] >= 3 and spacing_check['charge'] != peak['charge'] ): logger.info( f'Spacing override at m/z {peak["mz"]:.2f}: z={peak["charge"]} -> z={spacing_check["charge"]}' ) peak['charge'] = spacing_check['charge'] peak['confidence'] = max(0.5, spacing_check['confidence']) peak['method'] = 'spacing_override' # Convert to JSON-serializable format peaks_with_charge = [] for peak in detected_peaks: peaks_with_charge.append( { 'mz': float(peak['mz']), 'intensity': float(peak['intensity']), 'charge': int(peak['charge']) if peak['charge'] is not None else None, 'confidence': float(peak['confidence']), 'method': peak['method'], } ) # Return spectrum data with auto-detected peaks return jsonify( { 'spectrum': {'mz': mz_values.tolist(), 'intensity': intensity_values.tolist()}, 'num_points': len(mz_values), 'mz_range': [float(np.min(mz_values)), float(np.max(mz_values))], 'resolution': int(estimated_resolution), 'auto_detected_peaks': peaks_with_charge, # NEW: Send peak list with z values } ) except Exception as e: import traceback traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/load_sample', methods=['GET']) def load_sample() -> FlaskResponse: """Load sample spectrum file for demo/testing""" try: # Clear caches when new spectrum is loaded global _peak_analysis_cache, _isotope_pattern_cache _peak_analysis_cache.clear() _isotope_pattern_cache.clear() # Sample file path (in sample_data directory) sample_file = os.path.join(os.path.dirname(__file__), 'sample_data', 'GG208.txt') if not os.path.exists(sample_file): return jsonify({'error': 'Sample file not found in sample_data directory.'}), 404 # Read file content with open(sample_file, 'r') as f: content = f.read() # Parse spectrum mz_values, intensity_values = analyzer.parse_txt_spectrum(content) if len(mz_values) == 0: return jsonify({'error': 'No valid data found in sample file'}), 400 # Auto-detect resolution from spectrum estimated_resolution = analyzer.estimate_resolution(mz_values, intensity_values) logger.info(f'Auto-detected resolution: {estimated_resolution}') # AUTO-DETECT ALL PEAKS AND ASSIGN CHARGES logger.info('Detecting peak regions (isotope envelopes) and assigning charge states...') detected_peaks = detect_all_peaks_with_charge( mz_values, intensity_values, prominence=0.01, charge_range=(1, 10), method='combination', merge_gap=1.5 ) logger.info(f'Found {len(detected_peaks)} isotope envelopes with charge states') # Cross-check each peak's charge with direct spacing measurement for peak in detected_peaks: if peak['charge'] is not None: spacing_check = analyzer.detect_charge_state(mz_values, intensity_values, peak['mz'], window=3.0) if ( spacing_check['charge'] is not None and spacing_check['num_peaks'] >= 3 and spacing_check['charge'] != peak['charge'] ): logger.info( f'Spacing override at m/z {peak["mz"]:.2f}: z={peak["charge"]} -> z={spacing_check["charge"]}' ) peak['charge'] = spacing_check['charge'] peak['confidence'] = max(0.5, spacing_check['confidence']) peak['method'] = 'spacing_override' # Convert to JSON-serializable format peaks_with_charge = [] for peak in detected_peaks: peaks_with_charge.append( { 'mz': float(peak['mz']), 'intensity': float(peak['intensity']), 'charge': int(peak['charge']) if peak['charge'] is not None else None, 'confidence': float(peak['confidence']), 'method': peak['method'], } ) # Return spectrum data with auto-detected peaks return jsonify( { 'spectrum': {'mz': mz_values.tolist(), 'intensity': intensity_values.tolist()}, 'num_points': len(mz_values), 'mz_range': [float(np.min(mz_values)), float(np.max(mz_values))], 'resolution': int(estimated_resolution), 'auto_detected_peaks': peaks_with_charge, } ) except Exception as e: import traceback traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/download_sample', methods=['GET']) def download_sample() -> FlaskResponse: """Download sample spectrum file so users can see the data format""" try: sample_file = os.path.join(os.path.dirname(__file__), 'sample_data', 'GG208.txt') if not os.path.exists(sample_file): return jsonify({'error': 'Sample file not found'}), 404 return send_file(sample_file, as_attachment=True, download_name='sample_spectrum.txt', mimetype='text/plain') except Exception as e: import traceback traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/isotope_pattern', methods=['POST']) def get_isotope_pattern() -> FlaskResponse: """Generate isotope pattern for a given formula""" try: data = request.get_json() formula = data.get('formula') charge = int(data.get('charge', 1)) resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) if not formula: return jsonify({'error': 'No formula provided'}), 400 pattern = analyzer.generate_isotope_pattern(formula, charge, resolution) return jsonify(pattern) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/find_composition', methods=['POST']) def find_composition() -> FlaskResponse: """Search for a composition in the experimental spectrum""" try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] formula = data.get('formula') charge = int(data.get('charge', 1)) qcl = int(data.get('qcl', 0)) resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) spectrum_data = data.get('spectrum') peaks_data = data.get('peaks') adducts_input = data.get('adducts', []) if not formula: return jsonify({'error': 'No formula provided'}), 400 # Parse adducts and calculate total mass and charge total_adduct_mass = 0.0 total_adduct_charge = 0 if adducts_input: for adduct_item in adducts_input: adduct_name = adduct_item.get('name', '') adduct_count = int(adduct_item.get('count', 1)) inline_mass = adduct_item.get('mass') inline_charge = adduct_item.get('charge') if inline_mass is not None and inline_charge is not None: adduct_mass = float(inline_mass) adduct_charge = int(inline_charge) elif adduct_name in analyzer.adducts: adduct_mass, adduct_charge = analyzer.adducts[adduct_name] else: logger.warning(f"Adduct '{adduct_name}' not found in library, skipping") continue total_adduct_mass += adduct_mass * adduct_count total_adduct_charge += adduct_charge * adduct_count logger.debug( f'Adduct: {adduct_count}×{adduct_name}: mass={adduct_mass * adduct_count:.4f} Da, charge={adduct_charge * adduct_count:+d}' ) logger.info(f'Total adducts: mass={total_adduct_mass:.4f} Da, charge={total_adduct_charge:+d}') # Sanitize formula - remove any "mz" suffix that might have been accidentally appended original_formula = formula formula = formula.strip() if formula.endswith('mz'): logger.warning( f"Removing 'mz' suffix from formula. Original: '{original_formula}', Cleaned: '{formula[:-2]}'" ) formula = formula[:-2] logger.debug( f"Received formula for composition search: '{formula}', charge: {charge}, qcl: {qcl}, adduct_mass: {total_adduct_mass}, adduct_charge: {total_adduct_charge}" ) if not spectrum_data or not peaks_data: return jsonify({'error': 'No spectrum data available. Please upload a spectrum first.'}), 400 # Extract spectrum arrays mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) # Search for composition result = analyzer.find_composition_in_spectrum( formula, charge, qcl, mz_values, intensity_values, peaks_data, resolution, adduct_mass=total_adduct_mass, adduct_charge=total_adduct_charge, ) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 # Peak analysis cache - stores results by (peak_apex_mz, dna_sequence, resolution) _peak_analysis_cache: dict[tuple[float, str, int], dict[str, Any]] = {} @app.route('/analyze_region', methods=['POST']) @check_same_origin def analyze_region() -> FlaskResponse: """Analyze a clicked region - auto-detect charge state and find compositions""" # Rate limiting check if not check_rate_limit(request.remote_addr): return jsonify({'error': 'Rate limit exceeded. Please wait before making more requests.'}), 429 logger.info('ANALYZE_REGION v2.1 - Processing request') sys.stdout.flush() # Force immediate output try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] clicked_mz = float(data.get('clicked_mz')) detected_centroid = data.get('detected_centroid') # Frontend-calculated centroid spectrum_data = data.get('spectrum') dna_sequence = data.get('dna_sequence', '') custom_xna = data.get('custom_xna', None) # Custom XNA nucleotide data # For XNA mode, sequence is not required (we use total mass directly) if not custom_xna and not dna_sequence: return jsonify({'error': 'DNA sequence is required'}), 400 if not spectrum_data: return jsonify({'error': 'No spectrum data'}), 400 # Extract spectrum arrays mz_values = np.array(spectrum_data['mz']) intensity_values = np.array(spectrum_data['intensity']) # Define window around clicked point (±2 m/z) window = 2.0 mask = (mz_values >= clicked_mz - window) & (mz_values <= clicked_mz + window) region_mz = mz_values[mask] region_int = intensity_values[mask] if len(region_mz) < 5: return jsonify({'error': 'Insufficient data in clicked region'}), 400 # Find the actual peak maximum in this region max_idx = np.argmax(region_int) peak_mz = region_mz[max_idx] max_int = region_int[max_idx] # Check if charge was already detected from upload (frontend passes it) detected_charge = data.get('detected_charge', None) detected_charge_confidence = data.get('charge_confidence', None) detected_charge_method = data.get('charge_method', None) if detected_charge is not None: # Verify auto-detected charge with direct spacing measurement charge = int(detected_charge) charge_confidence = float(detected_charge_confidence) if detected_charge_confidence is not None else 0.8 charge_method = detected_charge_method if detected_charge_method is not None else 'auto_detected' # Cross-check: spacing method is more reliable for isotope-rich species (e.g. DNA-AgN) spacing_check = analyzer.detect_charge_state(mz_values, intensity_values, peak_mz, window=3.0) if spacing_check['charge'] is not None and spacing_check['num_peaks'] >= 3: spacing_charge = spacing_check['charge'] spacing_conf = float(spacing_check.get('confidence', 0.0)) if spacing_charge != charge: # Guard against domain-wrong spacing overrides: # (a) Ag-doublet halving artifact: spacing detector halves z because # Ag isotope doublets look like z/2 spacing — exactly the signal # we are trying to assign, so this halving is actively wrong here. # (b) Senko upload-time auto-detect with high confidence already # considered the full envelope; the single-peak spacing heuristic # is less reliable at high z (spacing < FWHM). halving_artifact = spacing_charge * 2 == charge trusted_supplied = charge_method == 'auto_detected' and charge_confidence >= 0.85 spacing_stronger = spacing_check['num_peaks'] >= 4 and spacing_conf > charge_confidence if halving_artifact or trusted_supplied or not spacing_stronger: logger.info( f'Spacing override SUPPRESSED at m/z {peak_mz:.4f}: spacing z={spacing_charge} ' f'(conf={spacing_conf:.2f}, peaks={spacing_check["num_peaks"]}) ' f'vs supplied z={charge} (conf={charge_confidence:.2f}, method={charge_method}); ' f'halving={halving_artifact}, trusted_supplied={trusted_supplied}, ' f'spacing_stronger={spacing_stronger}' ) else: logger.info( f'Spacing method disagrees: z={spacing_charge} (from {spacing_check["num_peaks"]} peaks, spacing={spacing_check["spacing"]:.4f}) vs auto-detected z={charge}' ) charge = spacing_charge charge_method = 'spacing_override' charge_confidence = max(0.5, spacing_conf) logger.info( f'Using charge for peak at m/z {peak_mz:.4f}: z={charge} (method: {charge_method}, confidence: {charge_confidence * 100:.1f}%)' ) else: # No auto-detected charge available - detect charge now # Use improved charge detection: # 1. Direct isotope spacing measurement (primary) # 2. Senko method fallback # 3. Prompt user for manual input if all fail logger.info(f'Detecting charge state for clicked peak at m/z {peak_mz:.4f}...') charge_result = analyzer.detect_charge_for_clicked_peak( mz_values, intensity_values, target_mz=peak_mz, charge_range=(1, 10) ) charge = charge_result['charge'] charge_confidence = charge_result['confidence'] charge_method = charge_result['method'] # If charge detection failed completely, return special response to prompt user if charge is None: logger.warning('Charge detection failed - prompting user for manual input') return jsonify( { 'charge_required': True, 'peak_mz': peak_mz, 'message': 'Could not automatically detect charge state. Please enter the charge (z) manually.', } ) logger.info( f'Final result: z={charge} (method: {charge_method}, confidence: {charge_confidence * 100:.1f}%)' ) # If confidence is very low (< 30%), warn user if charge_confidence < 0.3: logger.warning( "Low confidence charge detection! Peak may be weak, noisy, or overlapping. Consider using 'Re-analyze with new z' if results seem incorrect" ) # Get resolution before calculating compositions resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) # STEP 0.5: Snap to peak apex to ensure consistent analysis # Find the peak maximum within ±1 m/z of clicked position logger.debug(f'STEP 0.5: Snap to peak apex (clicked at {peak_mz:.4f})') snap_window = 1.0 snap_mask = (mz_values >= peak_mz - snap_window) & (mz_values <= peak_mz + snap_window) snap_mz = mz_values[snap_mask] snap_int = intensity_values[snap_mask] if len(snap_int) > 0: apex_idx = np.argmax(snap_int) peak_apex = float(snap_mz[apex_idx]) logger.debug(f'Peak apex found: {peak_apex:.4f} m/z (shift: {abs(peak_apex - peak_mz):.4f} m/z)') # Use apex as center for all subsequent analysis peak_mz = peak_apex else: logger.debug('No data in snap window, using clicked position') # Check cache - if we've analyzed this exact peak before, return cached result # Include custom_xna is_complex flag to avoid returning non-complex cached results in complex mode is_complex_for_cache = custom_xna.get('is_complex', False) if custom_xna else False cache_key = (round(peak_mz, 3), dna_sequence, resolution, is_complex_for_cache) # Round to 0.001 m/z precision if cache_key in _peak_analysis_cache: logger.debug(f'CACHE HIT! Returning cached result for peak {peak_mz:.4f} m/z') cached_data = _peak_analysis_cache[cache_key] try: result_json = jsonify(cached_data) return result_json except Exception as e: logger.error(f'Error serializing cached data: {e}', exc_info=True) # Clear bad cache entry and fall through to recompute del _peak_analysis_cache[cache_key] logger.debug('Cleared bad cache entry, will recompute') # STEP 1: Generate DISPLAY smooth envelope FIRST (this is what user sees) logger.debug(f'STEP 1: Generate DISPLAY smooth envelope (centered at {peak_mz:.4f})') window = 3.0 # ±3 m/z window for display (same as backup version) mask = (mz_values >= peak_mz - window) & (mz_values <= peak_mz + window) exp_mz_window = mz_values[mask] exp_int_window = intensity_values[mask] logger.debug( f'Using DISPLAY window: [{peak_mz - window:.4f}, {peak_mz + window:.4f}] m/z ({len(exp_mz_window)} points)' ) exp_gaussian_mz = [] exp_gaussian_intensity = [] exp_x0 = None exp_sigma = None if len(exp_mz_window) > 0: # Generate smooth Gaussian envelope exp_mz_gauss, exp_int_gauss = analyzer.generate_experimental_gaussian_envelope( exp_mz_window, exp_int_window, resolution ) if exp_mz_gauss is not None and exp_int_gauss is not None and len(exp_mz_gauss) > 3: # Check if envelope is truly flat (no isotope structure at all) mean_int_window = float(np.mean(exp_int_window)) mean_int_gauss = float(np.mean(exp_int_gauss)) raw_cv = float(np.std(exp_int_window)) / mean_int_window if mean_int_window > 0 else 0.0 envelope_cv = float(np.std(exp_int_gauss)) / mean_int_gauss if mean_int_gauss > 0 else 0.0 # Only flag as flat if the envelope itself has almost no variation (< 5%) is_envelope_flat = envelope_cv < 0.05 if is_envelope_flat: logger.debug( f'Envelope too flat (raw_cv={raw_cv * 100:.1f}%, envelope_cv={envelope_cv * 100:.1f}%) - not displaying envelope or X0' ) # Keep exp_gaussian_mz, exp_gaussian_intensity as empty lists # Keep exp_x0, exp_sigma as None else: # Fit Gaussian using gaussian_fit_centroid (same method as custom search) fit_result = analyzer.gaussian_fit_centroid(exp_mz_gauss, exp_int_gauss) if fit_result and fit_result[0] is not None: exp_x0 = fit_result[0] exp_sigma = fit_result[1] else: exp_x0, exp_sigma = None, None # Keep original envelope for display - it has the correct shape exp_gaussian_mz = exp_mz_gauss.tolist() exp_gaussian_intensity = exp_int_gauss.tolist() # Calculate peak symmetry symmetry_info = analyzer.calculate_peak_symmetry( mz_values=mz_values, intensity_values=intensity_values, center_mz=peak_mz, window=2.0 ) # STEP 3: Calculate compositions and filter using the CORRECT X₀ from narrow envelope # If envelope is flat, fallback to peak_mz if exp_x0 is None: exp_x0 = peak_mz logger.debug(f'STEP 3: Calculate compositions using X₀={exp_x0:.4f} (fallback to clicked peak)') else: logger.debug(f'STEP 3: Calculate compositions using X₀={exp_x0:.4f}') if custom_xna and custom_xna.get('formula'): # Use user-provided molecular weight if available, otherwise calculate from formula xna_mass = custom_xna.get('molecular_weight') if xna_mass is None: xna_mass = analyzer.calculate_mass_from_formula(custom_xna['formula']) logger.info( f'Using custom XNA: {custom_xna["name"]} (Formula: {custom_xna["formula"]}, Calculated Mass: {xna_mass:.2f} Da)' ) else: logger.info( f'Using custom XNA: {custom_xna["name"]} (Formula: {custom_xna["formula"]}, User-Provided Mass: {xna_mass:.2f} Da)' ) elif custom_xna and custom_xna.get('is_complex'): # DNA-only Complex mode - no XNA formula, will use DNA sequence for mass calculation logger.info( f'Complex DNA mode (no XNA formula) - is_complex={custom_xna.get("is_complex")}, using DNA sequence for mass calculation' ) # Use smart adduct search (mass-based filtering, triggered by X₀ error > 0.5) compositions = analyzer.analyze_peak_with_smart_adduct_search( peak_mz, charge, dna_sequence, exp_x0, resolution=resolution, mz_values=mz_values, intensity_values=intensity_values, custom_xna=custom_xna, ) # Refine with isotope matching - this will filter based on CORRECT X₀ has_other_strands = False all_compositions = [] if len(compositions) > 0: ( compositions, exp_x0_refined, exp_sigma_refined, has_other_strands, all_compositions, has_odd_n0_warning, _, _, has_unrealistic_n0_warning, ) = analyzer.refine_compositions_with_isotope_matching( compositions, mz_values, intensity_values, peak_mz, resolution=resolution, detected_centroid=exp_x0, # Use DISPLAY envelope X₀ ) # Keep the display envelope X₀ (don't overwrite with narrow envelope) # exp_x0 stays as the display envelope value else: has_odd_n0_warning = False has_unrealistic_n0_warning = False # Calculate composition estimates (for when no auto compositions found) composition_estimates = [] is_complex = custom_xna.get('is_complex', False) if custom_xna else False if len(compositions) == 0 and charge is not None: composition_estimates = analyzer.calculate_composition_estimates( peak_mz, charge, dna_sequence, custom_xna=custom_xna, max_strands=MAX_STRANDS * 2, is_complex=is_complex ) logger.debug( f'Calculated {len(composition_estimates)} composition estimates for user guidance (complex={is_complex})' ) # Convert all NumPy types to native Python types for JSON serialization result = { 'clicked_mz': clicked_mz, 'peak_mz': float(peak_mz), 'intensity': float(max_int), 'charge': charge, 'charge_method': charge_method, 'charge_confidence': float(charge_confidence) if charge_confidence is not None else None, 'compositions': compositions, 'composition_estimates': composition_estimates, 'region_mz_range': [float(region_mz[0]), float(region_mz[-1])], 'exp_x0': float(exp_x0) if exp_x0 is not None else None, 'exp_sigma': float(exp_sigma) if exp_sigma is not None else None, 'symmetry': symmetry_info, 'has_other_strands': has_other_strands, 'all_compositions': all_compositions, 'has_odd_n0_warning': has_odd_n0_warning, 'has_unrealistic_n0_warning': has_unrealistic_n0_warning, 'exp_gaussian_mz': exp_gaussian_mz, 'exp_gaussian_intensity': exp_gaussian_intensity, 'is_complex': is_complex, } logger.debug(f'Sending response: exp_gaussian_mz={len(exp_gaussian_mz)} points') # Convert to JSON-serializable format and cache the result json_result = convert_numpy_types(result) _peak_analysis_cache[cache_key] = json_result logger.debug(f'Cached result for peak {peak_mz:.4f} m/z (cache size: {len(_peak_analysis_cache)})') return jsonify(json_result) except Exception as e: logger.error( f'ANALYSIS_CRASH in analyze_region: Input={data.get("clicked_mz", "unknown")}, Error={type(e).__name__}: {str(e)}', exc_info=True, ) return jsonify({'error': str(e)}), 500 @app.route('/analyze_peak', methods=['POST']) @check_same_origin def analyze_peak() -> FlaskResponse: """Analyze a specific peak for composition (legacy endpoint)""" # Rate limiting check if not check_rate_limit(request.remote_addr): return jsonify({'error': 'Rate limit exceeded. Please wait before making more requests.'}), 429 try: data = request.get_json() analyzer = DNASilverAnalyzer() analyzer.custom_adducts = data.get('custom_adducts', []) or [] mz = float(data.get('mz')) charge = int(data.get('charge', 1)) dna_sequence = data.get('dna_sequence', None) resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) # For now, without neighboring peaks, assume charge from input compositions = analyzer.calculate_dna_silver_composition(mz, charge, dna_sequence, resolution=resolution) return jsonify({'mz': mz, 'charge': charge, 'compositions': compositions}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/get_all_adducts', methods=['GET']) def get_all_adducts() -> FlaskResponse: """Get list of ALL adducts (built-in + custom) from the library""" try: # Build list of all adducts with their properties all_adducts = [] for name, (mass, charge) in analyzer.adducts.items(): all_adducts.append({'name': name, 'mass': mass, 'charge': charge}) # Sort by name for easier selection all_adducts.sort(key=lambda x: x['name']) return jsonify({'adducts': all_adducts, 'total_count': len(all_adducts)}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/parse_adduct_formula', methods=['POST']) def parse_adduct_formula() -> FlaskResponse: """Stateless: validate name + formula/mass + charge and return {mass, charge}. Custom adducts live in the client's localStorage, not on the server, so different HF Spaces users never share or overwrite each other's lists. """ try: data = request.get_json() name = (data.get('name') or '').strip() formula = (data.get('formula') or '').strip() mass = data.get('mass') charge = data.get('charge') if not name: return jsonify({'success': False, 'error': 'Adduct name is required'}), 400 if charge is None: return jsonify({'success': False, 'error': 'Charge is required'}), 400 try: charge = int(charge) except (TypeError, ValueError): return jsonify({'success': False, 'error': 'Charge must be an integer'}), 400 if not formula and mass in (None, ''): return jsonify({'success': False, 'error': 'Formula or mass is required'}), 400 computed_mass = None resolved_formula = None if formula: ok, calc_mass, err = analyzer.calculate_mass_from_formula_with_validation(formula) if ok: computed_mass = calc_mass resolved_formula = formula else: try: computed_mass = float(formula) except ValueError: return jsonify({'success': False, 'error': err or f'Invalid formula: {formula}'}), 400 else: try: computed_mass = float(mass) except (TypeError, ValueError): return jsonify({'success': False, 'error': 'Invalid mass value'}), 400 if computed_mass is None or computed_mass < 0.1 or computed_mass > 10000: return jsonify({'success': False, 'error': 'Mass must be between 0.1 and 10000 Da'}), 400 if charge < -5 or charge > 5: return jsonify({'success': False, 'error': 'Charge must be between -5 and +5'}), 400 return jsonify( { 'success': True, 'name': name, 'formula': resolved_formula, 'mass': computed_mass, 'charge': charge, } ) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/smiles_to_formula', methods=['POST']) def smiles_to_formula() -> FlaskResponse: """Convert SMILES string to molecular formula using RDKit""" try: data = request.get_json() smiles = data.get('smiles', '').strip() if not smiles: return jsonify({'error': 'No SMILES string provided'}), 400 try: from rdkit import Chem from rdkit.Chem import rdMolDescriptors # Parse SMILES mol = Chem.MolFromSmiles(smiles) if mol is None: return jsonify({'error': f'Invalid SMILES string: {smiles}'}), 400 # Add hydrogens to get complete formula mol = Chem.AddHs(mol) # Get molecular formula formula = rdMolDescriptors.CalcMolFormula(mol) # Get molecular weight for reference mol_weight = rdMolDescriptors.CalcExactMolWt(mol) return jsonify({'formula': formula, 'smiles': smiles, 'molecular_weight': mol_weight}) except ImportError: # RDKit not installed - provide fallback message return jsonify({'error': 'RDKit is not installed. Please install it with: pip install rdkit'}), 500 except Exception as e: logger.error(f'Error in smiles_to_formula: {e}', exc_info=True) return jsonify({'error': str(e)}), 500 @app.route('/isotope_library', methods=['GET', 'POST']) def isotope_library_settings() -> FlaskResponse: """ GET: Return current isotope library settings POST: Switch isotope library (isospec or pythoms) """ global ISOTOPE_LIBRARY, _isotope_pattern_cache if request.method == 'GET': return jsonify( { 'current_library': ISOTOPE_LIBRARY, 'isospec_available': ISOSPEC_AVAILABLE, 'cache_size': len(_isotope_pattern_cache), } ) # POST - switch library try: data = request.get_json() new_library = data.get('library', '').lower() if new_library not in ['isospec', 'pythoms']: return jsonify({'error': 'Invalid library. Use "isospec" or "pythoms"'}), 400 if new_library == 'isospec' and not ISOSPEC_AVAILABLE: return jsonify({'error': 'IsoSpecPy is not installed. Install with: pip install IsoSpecPy'}), 400 old_library = ISOTOPE_LIBRARY ISOTOPE_LIBRARY = new_library # Clear cache when switching libraries to ensure consistency cache_size = len(_isotope_pattern_cache) _isotope_pattern_cache.clear() logger.info(f'Switched isotope library: {old_library} → {new_library} (cleared {cache_size} cached patterns)') return jsonify( {'success': True, 'old_library': old_library, 'new_library': new_library, 'cache_cleared': cache_size} ) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/benchmark_isotope', methods=['POST']) def benchmark_isotope() -> FlaskResponse: """ Benchmark isotope pattern generation with both libraries. Returns timing comparison. """ import time try: data = request.get_json() formula = data.get('formula', 'C200H280N80O120P20Ag16') charge = int(data.get('charge', 4)) resolution = int(data.get('resolution', DEFAULT_RESOLUTION)) iterations = int(data.get('iterations', 5)) benchmark_analyzer = DNASilverAnalyzer() results: dict[str, Any] = {} # Clear cache before benchmarking global _isotope_pattern_cache _isotope_pattern_cache.clear() # Benchmark PythoMS start = time.time() for _ in range(iterations): _isotope_pattern_cache.clear() # Clear cache each iteration pattern_pythoms = benchmark_analyzer._generate_isotope_pattern_pythoms(formula, charge, resolution) pythoms_time = (time.time() - start) / iterations results['pythoms'] = {'time_per_call': pythoms_time, 'success': 'error' not in pattern_pythoms} # Benchmark IsoSpecPy if available if ISOSPEC_AVAILABLE: start = time.time() for _ in range(iterations): _isotope_pattern_cache.clear() pattern_isospec = benchmark_analyzer._generate_isotope_pattern_isospec(formula, charge, resolution) isospec_time = (time.time() - start) / iterations results['isospec'] = {'time_per_call': isospec_time, 'success': 'error' not in pattern_isospec} if results['pythoms']['success'] and results['isospec']['success']: results['speedup'] = pythoms_time / isospec_time # Compare X0 values pythoms_mz = np.array(pattern_pythoms['mz']) pythoms_int = np.array(pattern_pythoms['intensity']) isospec_mz = np.array(pattern_isospec['mz']) isospec_int = np.array(pattern_isospec['intensity']) pythoms_x0 = np.average(pythoms_mz, weights=pythoms_int) isospec_x0 = np.average(isospec_mz, weights=isospec_int) results['x0_comparison'] = { 'pythoms_x0': pythoms_x0, 'isospec_x0': isospec_x0, 'difference': abs(pythoms_x0 - isospec_x0), } else: results['isospec'] = {'error': 'IsoSpecPy not installed'} results['formula'] = formula results['charge'] = charge results['iterations'] = iterations return jsonify(convert_numpy_types(results)) except Exception as e: logger.error(f'Error in benchmark_isotope: {e}', exc_info=True) return jsonify({'error': str(e)}), 500 if __name__ == '__main__': # Create templates directory if it doesn't exist os.makedirs('templates', exist_ok=True) # Check if port 8080 is available import socket def is_port_in_use(port: int) -> bool: """Check if a port is already in use""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: try: sock.settimeout(1) result = sock.connect_ex(('localhost', port)) return result == 0 # True if port is in use except Exception: return False port = int(os.environ.get('PORT', 8080)) logger.info('Checking for existing server instances...') if is_port_in_use(port): logger.warning( f'Port {port} is already in use! Another server instance may be running. Please stop it first or use a different port.' ) import sys sys.exit(1) else: logger.info('No existing instances found') # Debug mode: OFF by default for security. Enable for development only: # export FLASK_DEBUG=1 (on Mac/Linux) # set FLASK_DEBUG=1 (on Windows) debug_mode = os.environ.get('FLASK_DEBUG', '0').lower() in ('1', 'true', 'yes') logger.info('Starting DNA-stabilized Silver Nanocluster Mass Spec Analysis Web Server...') logger.info(f'Open your browser to http://localhost:{port}') if debug_mode: logger.warning('Debug mode: ON (for development only, not secure for production)') else: logger.info('Debug mode: OFF (production-safe)') logger.info('Press CTRL+C to stop the server') # Enable threaded mode and use_reloader=False to avoid port conflicts app.run(debug=debug_mode, host='0.0.0.0', port=port, threaded=True, use_reloader=False)