Spaces:
Running
Running
| """ | |
| 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 | |
| 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() | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| def index() -> str: | |
| """Render main page""" | |
| return render_template('index.html') | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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]] = {} | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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) | |