viviandlin commited on
Commit
98c5587
·
1 Parent(s): 9ecdbfb

Sync from GitHub: Refactor analyzer into mixin modules for maintainability

Browse files

Split monolithic core/analyzer.py into focused mixin modules
(spectrum, envelope, scoring, isotope, charge) composed via MRO.
Deduplicate adduct parser in webapp, fix typo in index.html.
No algorithmic changes; all validation results unchanged.

.last_sync CHANGED
@@ -1 +1 @@
1
- 7ae1298887e7521bb82fd86e504071b599dbe48c
 
1
+ 1ba9fe091732330384694b2f4d259c9efde1f1ae
README.md CHANGED
@@ -108,12 +108,6 @@ Supported formats: .txt, .csv
108
  - **Frontend:** HTML5, JavaScript, Plotly.js, JSME
109
  - **Libraries:** PythoMS, IsoSpecPy
110
 
111
- ## Citation
112
-
113
- If you use NucleoSpec in a publication, please cite:
114
-
115
- > Lin, I.-H.; Copp, S. M. A Tutorial on Automated Mass Spectral Analysis using NucleoSpec for Compositional Assignment of Nucleic Acid–Silver Complexes and Nanoclusters. *ChemRxiv* 2026. [DOI: 10.26434/chemrxiv.15004738/v1](https://doi.org/10.26434/chemrxiv.15004738/v1)
116
-
117
  ## Support
118
 
119
  - **Developer:** I-Hsin (Vivian) Lin
 
108
  - **Frontend:** HTML5, JavaScript, Plotly.js, JSME
109
  - **Libraries:** PythoMS, IsoSpecPy
110
 
 
 
 
 
 
 
111
  ## Support
112
 
113
  - **Developer:** I-Hsin (Vivian) Lin
core/charge.py ADDED
@@ -0,0 +1,227 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import logging
4
+ import os
5
+ import sys
6
+ from typing import Optional
7
+
8
+ import numpy as np
9
+ import numpy.typing as npt
10
+
11
+ current_dir = os.path.dirname(os.path.abspath(__file__))
12
+ sys.path.insert(0, os.path.join(current_dir, '..', 'lib'))
13
+
14
+
15
+ logger = logging.getLogger(__name__)
16
+
17
+
18
+ class ChargeMixin:
19
+ """Mixin for charge state detection from isotope spacing."""
20
+
21
+ def group_isotope_envelope(
22
+ self, peak_mz: npt.NDArray[np.float64], peak_intensity: npt.NDArray[np.float64], charge: Optional[int]
23
+ ) -> Optional[int]:
24
+ """
25
+ Group peaks that belong to the same isotope envelope
26
+ Returns the index of the most intense peak (representative peak)
27
+ """
28
+ if charge is None or charge <= 0:
29
+ return None
30
+
31
+ # Expected spacing for this charge state
32
+ spacing = 1.003 / charge
33
+
34
+ # Find the most intense peak in this envelope
35
+ return int(np.argmax(peak_intensity))
36
+
37
+ def detect_charge_state(
38
+ self,
39
+ mz_values: npt.NDArray[np.float64],
40
+ intensity_values: npt.NDArray[np.float64],
41
+ target_mz: float,
42
+ window: float = 3.0,
43
+ ) -> dict:
44
+ """
45
+ Detect charge state by anchored isotope-grid scoring.
46
+
47
+ For each candidate z in [1..10], build the expected isotope grid
48
+ target_mz + k * (1.003/z) and score how well the experimental peaks fit:
49
+ score = (fraction of strong-peak intensity on the grid)
50
+ - (fraction of grid positions with no peak nearby)
51
+ Best z = highest score among viable candidates. If best is even and the
52
+ grid-intensity pattern alternates high-low (Ag doublet for DNA-AgN
53
+ clusters), halve z — this distinguishes a true z=N envelope from a
54
+ z=N/2 envelope where 107Ag/109Ag doubles the apparent peak count.
55
+
56
+ Returns dict compatible with previous callers:
57
+ 'spacing', 'charge', 'confidence', 'num_peaks', 'scores'
58
+ """
59
+ from scipy.signal import find_peaks
60
+
61
+ NEUTRON_MASS = 1.003
62
+ CHARGE_RANGE = (1, 10)
63
+ DOUBLET_ALT_THRESHOLD = 0.85
64
+
65
+ mask = (mz_values >= target_mz - window) & (mz_values <= target_mz + window)
66
+ region_mz = mz_values[mask]
67
+ region_int = intensity_values[mask]
68
+
69
+ if len(region_mz) < 5:
70
+ return {'spacing': None, 'charge': None, 'confidence': 0.0, 'num_peaks': 0, 'scores': {}}
71
+
72
+ max_intensity = float(np.max(region_int))
73
+ peaks_idx, _ = find_peaks(region_int, prominence=max_intensity * 0.03, distance=1)
74
+
75
+ if len(peaks_idx) < 2:
76
+ return {'spacing': None, 'charge': None, 'confidence': 0.0, 'num_peaks': int(len(peaks_idx)), 'scores': {}}
77
+
78
+ peak_mzs = region_mz[peaks_idx]
79
+ peak_ints = region_int[peaks_idx]
80
+
81
+ strong_mask = peak_ints >= max_intensity * 0.10
82
+ strong_mzs = peak_mzs[strong_mask]
83
+ strong_ints = peak_ints[strong_mask]
84
+ total_strong_int = float(np.sum(strong_ints)) if len(strong_ints) > 0 else 1.0
85
+
86
+ results: dict[int, dict[str, float]] = {}
87
+ for z in range(CHARGE_RANGE[0], CHARGE_RANGE[1] + 1):
88
+ spacing = NEUTRON_MASS / z
89
+ tol = spacing * 0.25
90
+ n_iso = max(1, int(window / spacing))
91
+ ks = np.arange(-n_iso, n_iso + 1)
92
+ grid = target_mz + ks * spacing
93
+
94
+ grid_ints = np.zeros(len(grid))
95
+ for i, g in enumerate(grid):
96
+ d = np.abs(peak_mzs - g)
97
+ j = int(np.argmin(d))
98
+ if d[j] <= tol:
99
+ grid_ints[i] = peak_ints[j]
100
+
101
+ matched_int = 0.0
102
+ for smz, sint in zip(strong_mzs, strong_ints):
103
+ if np.min(np.abs(grid - smz)) <= tol:
104
+ matched_int += float(sint)
105
+ coverage = matched_int / total_strong_int
106
+ gap_frac = float(np.sum(grid_ints == 0)) / len(grid)
107
+
108
+ even_vals = grid_ints[(ks % 2 == 0) & (grid_ints > 0)]
109
+ odd_vals = grid_ints[(ks % 2 == 1) & (grid_ints > 0)]
110
+ if len(even_vals) > 0 and len(odd_vals) > 0:
111
+ em, om = float(even_vals.mean()), float(odd_vals.mean())
112
+ alt = (min(em, om) / max(em, om)) if max(em, om) > 0 else 1.0
113
+ else:
114
+ alt = 1.0
115
+
116
+ left = sum(1 for k in range(1, n_iso + 1) if np.min(np.abs(peak_mzs - (target_mz - k * spacing))) <= tol)
117
+ right = sum(1 for k in range(1, n_iso + 1) if np.min(np.abs(peak_mzs - (target_mz + k * spacing))) <= tol)
118
+
119
+ results[z] = {
120
+ 'score': float(coverage - gap_frac),
121
+ 'coverage': float(coverage),
122
+ 'gap_frac': float(gap_frac),
123
+ 'alt': float(alt),
124
+ 'left': int(left),
125
+ 'right': int(right),
126
+ 'viable': bool(left + right + 1 >= 5),
127
+ 'spacing': float(spacing),
128
+ }
129
+
130
+ viable_zs = [z for z, r in results.items() if r['viable']]
131
+ if viable_zs:
132
+ best_z = max(viable_zs, key=lambda z: results[z]['score'])
133
+ else:
134
+ best_z = max(results.keys(), key=lambda z: results[z]['score'])
135
+
136
+ while best_z > 1 and best_z % 2 == 0:
137
+ half = best_z // 2
138
+ if half in results and results[half]['viable'] and results[best_z]['alt'] < DOUBLET_ALT_THRESHOLD:
139
+ logger.info(
140
+ f'[detect_charge_state] Ag-doublet halving at m/z {target_mz:.4f}: '
141
+ f'z={best_z} -> z={half} (alt={results[best_z]["alt"]:.2f})'
142
+ )
143
+ best_z = half
144
+ else:
145
+ break
146
+
147
+ best = results[best_z]
148
+ confidence = max(0.0, min(1.0, best['score']))
149
+ num_matched = best['left'] + best['right'] + 1
150
+
151
+ logger.debug(
152
+ f'[detect_charge_state] target={target_mz:.4f} -> z={best_z} '
153
+ f'(coverage={best["coverage"]:.2f}, gap={best["gap_frac"]:.2f}, '
154
+ f'alt={best["alt"]:.2f}, score={best["score"]:+.3f})'
155
+ )
156
+
157
+ return {
158
+ 'spacing': float(best['spacing']),
159
+ 'charge': int(best_z),
160
+ 'confidence': float(confidence),
161
+ 'num_peaks': int(num_matched),
162
+ 'scores': results,
163
+ }
164
+
165
+ def detect_charge_for_clicked_peak(
166
+ self,
167
+ mz_values: npt.NDArray[np.float64],
168
+ intensity_values: npt.NDArray[np.float64],
169
+ target_mz: float,
170
+ charge_range: tuple[int, int] = (1, 10),
171
+ ) -> dict:
172
+ """
173
+ Determine charge state of a user-clicked peak.
174
+
175
+ Primary: isotope-grid scoring via detect_charge_state.
176
+ Fallback: Senko charge assignment on the surrounding envelope.
177
+ Returns dict with 'charge', 'confidence', 'method' (and 'spacing',
178
+ 'num_peaks' when produced by the primary method). 'charge' is None
179
+ only when both methods fail.
180
+ """
181
+ logger.info(f'[Charge Detection] Analyzing peak at m/z {target_mz:.4f}')
182
+
183
+ result = self.detect_charge_state(mz_values, intensity_values, target_mz, window=3.0)
184
+ charge = result.get('charge')
185
+ if charge is not None and charge_range[0] <= charge <= charge_range[1]:
186
+ num_peaks = int(result.get('num_peaks', 0))
187
+ confidence = float(result.get('confidence', 0.0))
188
+ if num_peaks < 3:
189
+ confidence = min(0.6, confidence)
190
+ logger.info(f'[Charge Detection] z={charge} via grid (conf={confidence * 100:.0f}%, {num_peaks} matched)')
191
+ return {
192
+ 'charge': int(charge),
193
+ 'confidence': float(confidence),
194
+ 'method': 'spacing',
195
+ 'spacing': float(result['spacing']),
196
+ 'num_peaks': num_peaks,
197
+ }
198
+
199
+ logger.debug('[Charge Detection] Grid method inconclusive, falling back to Senko')
200
+ try:
201
+ from pythoms.senko_charge_assignment import detect_all_peaks_with_charge
202
+
203
+ detected = detect_all_peaks_with_charge(
204
+ mz_values,
205
+ intensity_values,
206
+ prominence=0.01,
207
+ charge_range=charge_range,
208
+ method='combination',
209
+ merge_gap=1.5,
210
+ )
211
+ closest = min(
212
+ (p for p in detected if abs(p['mz'] - target_mz) < 5.0 and p.get('charge') is not None),
213
+ key=lambda p: abs(p['mz'] - target_mz),
214
+ default=None,
215
+ )
216
+ if closest is not None:
217
+ logger.info(f'[Charge Detection] z={closest["charge"]} via Senko fallback')
218
+ return {
219
+ 'charge': int(closest['charge']),
220
+ 'confidence': float(closest['confidence']) * 0.8,
221
+ 'method': 'senko_fallback',
222
+ }
223
+ except Exception as e:
224
+ logger.error(f'[Charge Detection] Senko fallback error: {e}')
225
+
226
+ logger.warning('[Charge Detection] All methods failed; user input required')
227
+ return {'charge': None, 'confidence': 0.0, 'method': 'user_input_required'}
core/envelope.py ADDED
@@ -0,0 +1,814 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import logging
4
+ from typing import Optional
5
+
6
+ import numpy as np
7
+ import numpy.typing as npt
8
+ from scipy.optimize import curve_fit
9
+
10
+ logger = logging.getLogger(__name__)
11
+
12
+
13
+ class EnvelopeMixin:
14
+ """Mixin for Gaussian envelope generation, fitting, and peak symmetry analysis."""
15
+
16
+ def smooth_gaussian_pattern(self, barip: list[list], fwhm: float, num_points_per_fwhm: int = 100) -> list[list]:
17
+ """
18
+ Generate smooth Gaussian isotope pattern with guaranteed high resolution sampling.
19
+
20
+ Args:
21
+ barip: Bar isotope pattern [[mz_values], [intensities]]
22
+ fwhm: Full width at half maximum
23
+ num_points_per_fwhm: Number of sampling points per FWHM (default: 100 for very smooth curves)
24
+
25
+ Returns:
26
+ [[mz_values], [intensities]] - smoothed Gaussian pattern
27
+ """
28
+ if not barip or len(barip[0]) == 0:
29
+ return [[], []]
30
+
31
+ mz_bar = np.array(barip[0])
32
+ int_bar = np.array(barip[1])
33
+
34
+ # Define m/z range: extend ±3*FWHM from min/max peaks (covers 99.7% of Gaussian)
35
+ mz_min = np.min(mz_bar) - 3 * fwhm
36
+ mz_max = np.max(mz_bar) + 3 * fwhm
37
+
38
+ # Calculate step size for smooth curve: FWHM / num_points_per_fwhm
39
+ step = fwhm / num_points_per_fwhm
40
+
41
+ # Generate high-resolution m/z grid
42
+ mz_grid = np.arange(mz_min, mz_max + step, step)
43
+ intensity_grid = np.zeros_like(mz_grid)
44
+
45
+ # Sigma (standard deviation) from FWHM: FWHM = 2.355 * sigma
46
+ sigma = fwhm / 2.355
47
+
48
+ # Sum Gaussian peaks for each isotope
49
+ for center_mz, height in zip(mz_bar, int_bar):
50
+ # Generate normalized Gaussian: exp(-(x-center)^2 / (2*sigma^2))
51
+ gaussian_contrib = np.exp(-0.5 * ((mz_grid - center_mz) / sigma) ** 2)
52
+ # Scale by height
53
+ intensity_grid += gaussian_contrib * height
54
+
55
+ # Normalize to 100
56
+ if np.max(intensity_grid) > 0:
57
+ intensity_grid = (intensity_grid / np.max(intensity_grid)) * 100.0
58
+
59
+ return [mz_grid.tolist(), intensity_grid.tolist()]
60
+
61
+ def calculate_peak_symmetry(
62
+ self,
63
+ mz_values: npt.NDArray[np.float64],
64
+ intensity_values: npt.NDArray[np.float64],
65
+ center_mz: float,
66
+ window: float = 2.0,
67
+ ) -> dict:
68
+ """
69
+ Calculate symmetry of a peak around its center.
70
+ Returns symmetry score (0-1, where 1 is perfectly symmetric)
71
+ and skewness indicator.
72
+
73
+ A symmetric peak suggests a clean, single species (like a nanocluster).
74
+ An asymmetric peak may indicate fragmentation, impurities, or overlapping peaks.
75
+ """
76
+ # Extract region around peak
77
+ mask = (mz_values >= center_mz - window) & (mz_values <= center_mz + window)
78
+ region_mz = mz_values[mask]
79
+ region_int = intensity_values[mask]
80
+
81
+ if len(region_mz) < 5:
82
+ return {'symmetry_score': 0.0, 'skewness': 0.0, 'is_symmetric': False, 'note': 'Insufficient data points'}
83
+
84
+ # Find peak apex
85
+ max_idx = np.argmax(region_int)
86
+ apex_mz = region_mz[max_idx]
87
+ max_intensity = region_int[max_idx]
88
+
89
+ # Divide into left and right sides from apex
90
+ left_mz = region_mz[: max_idx + 1]
91
+ left_int = region_int[: max_idx + 1]
92
+ right_mz = region_mz[max_idx:]
93
+ right_int = region_int[max_idx:]
94
+
95
+ if len(left_mz) < 2 or len(right_mz) < 2:
96
+ return {'symmetry_score': 0.0, 'skewness': 0.0, 'is_symmetric': False, 'note': 'Peak too narrow'}
97
+
98
+ # Calculate statistical skewness
99
+ mean_mz = np.average(region_mz, weights=region_int)
100
+ variance = np.average((region_mz - mean_mz) ** 2, weights=region_int)
101
+ std_dev = np.sqrt(variance)
102
+
103
+ if std_dev > 0:
104
+ skewness = np.average(((region_mz - mean_mz) / std_dev) ** 3, weights=region_int)
105
+ else:
106
+ skewness = 0.0
107
+
108
+ # Compare left and right sides by mirroring around apex
109
+ max_distance = min(apex_mz - region_mz[0], region_mz[-1] - apex_mz)
110
+
111
+ symmetry_scores = []
112
+ symmetry_weights = []
113
+ num_points = min(20, int(max_distance / 0.05)) # Finer sampling for better accuracy
114
+
115
+ for i in range(1, num_points + 1):
116
+ offset = (i / num_points) * max_distance
117
+
118
+ # Find intensity at left and right positions
119
+ left_pos = apex_mz - offset
120
+ right_pos = apex_mz + offset
121
+
122
+ # Interpolate intensities
123
+ left_intensity = np.interp(left_pos, left_mz, left_int, left=0, right=0)
124
+ right_intensity = np.interp(right_pos, right_mz, right_int, left=0, right=0)
125
+
126
+ # Calculate local symmetry, weighted by average intensity
127
+ # so high-signal regions near apex matter more than low-signal tails
128
+ avg_intensity = (left_intensity + right_intensity) / 2.0
129
+ if avg_intensity > 0:
130
+ local_asym = abs(left_intensity - right_intensity) / (left_intensity + right_intensity)
131
+ symmetry_scores.append(1.0 - local_asym)
132
+ symmetry_weights.append(avg_intensity)
133
+
134
+ # Overall symmetry score (intensity-weighted average)
135
+ if symmetry_scores and symmetry_weights:
136
+ symmetry_score = float(np.average(symmetry_scores, weights=symmetry_weights))
137
+ else:
138
+ symmetry_score = 0.0
139
+
140
+ # Determine if peak is symmetric
141
+ is_symmetric = symmetry_score > 0.7 and abs(skewness) < 0.5
142
+
143
+ # Generate interpretation
144
+ if symmetry_score > 0.85 and abs(skewness) < 0.3:
145
+ note = 'Highly symmetric - likely clean nanocluster'
146
+ elif symmetry_score > 0.7 and abs(skewness) < 0.5:
147
+ note = 'Moderately symmetric - good quality'
148
+ elif symmetry_score > 0.5:
149
+ note = 'Slightly asymmetric - may have impurities'
150
+ else:
151
+ note = 'Asymmetric - possible fragmentation or overlapping peaks'
152
+
153
+ return {
154
+ 'symmetry_score': float(symmetry_score),
155
+ 'skewness': float(skewness),
156
+ 'is_symmetric': bool(is_symmetric), # Convert numpy bool to Python bool
157
+ 'note': note,
158
+ 'apex_mz': float(apex_mz),
159
+ }
160
+
161
+ def generate_experimental_gaussian_envelope(
162
+ self, exp_mz: npt.NDArray[np.float64], exp_int: npt.NDArray[np.float64], resolution: int
163
+ ) -> tuple[Optional[npt.NDArray], Optional[npt.NDArray]]:
164
+ """
165
+ Generate smooth Gaussian envelope for experimental data.
166
+ Uses Gaussian smoothing with kernel based on instrument resolution.
167
+ This will show the natural asymmetry of the experimental data.
168
+ """
169
+ try:
170
+ logger.debug('GENERATE_EXPERIMENTAL_GAUSSIAN_ENVELOPE CALLED')
171
+ logger.debug(f'Input: {len(exp_mz)} m/z points, resolution={resolution}')
172
+
173
+ if len(exp_mz) == 0 or len(exp_int) == 0:
174
+ logger.warning('FAILED: Empty input data')
175
+ return None, None
176
+
177
+ # Convert to numpy arrays
178
+ exp_mz = np.array(exp_mz)
179
+ exp_int = np.array(exp_int)
180
+
181
+ # Calculate FWHM and sigma from resolution
182
+ peak_center = np.average(exp_mz, weights=exp_int)
183
+ fwhm = peak_center / resolution
184
+ sigma = fwhm / 2.355 # Convert FWHM to sigma
185
+
186
+ logger.debug(f'Peak center: {peak_center:.4f}, FWHM: {fwhm:.6f}, sigma: {sigma:.6f}')
187
+
188
+ # SMART APPROACH: Find apex (local maximum) of each isotope peak
189
+ # Then use the SAME smooth_gaussian_pattern function as theoretical data
190
+ # This ensures consistent smooth curves!
191
+
192
+ from scipy.signal import find_peaks
193
+
194
+ # Find local maxima (apex of each isotope peak)
195
+ # Use a small distance to separate isotope peaks (~0.2 Da for typical spacing)
196
+ min_distance = int(0.2 / np.median(np.diff(exp_mz))) if len(exp_mz) > 1 else 2
197
+ peaks_idx, properties = find_peaks(
198
+ exp_int, distance=max(2, min_distance), prominence=np.max(exp_int) * 0.05
199
+ )
200
+
201
+ if len(peaks_idx) < 3:
202
+ # Not enough peaks found - use all data points
203
+ logger.debug(f'Found only {len(peaks_idx)} apex points, using all data')
204
+ apex_mz = exp_mz
205
+ apex_int = exp_int
206
+ else:
207
+ # Extract apex points
208
+ all_apex_mz = exp_mz[peaks_idx]
209
+ all_apex_int = exp_int[peaks_idx]
210
+ logger.debug(f'Found {len(all_apex_mz)} apex points (local maxima)')
211
+
212
+ # FILTER: keep apex points that form a contiguous series with the
213
+ # most-intense apex. Walk left/right until a gap larger than
214
+ # 2.5 × median isotope spacing is encountered (the next envelope).
215
+ # This adapts to envelope width — narrow at low mass, wider at high
216
+ # mass where many Ag atoms broaden the isotope distribution.
217
+ max_apex_idx = int(np.argmax(all_apex_int))
218
+ spacings = np.diff(all_apex_mz)
219
+ median_spacing = float(np.median(spacings)) if len(spacings) >= 1 else 0.334
220
+ gap_threshold = max(median_spacing * 3.0, 0.5)
221
+
222
+ start = max_apex_idx
223
+ end = max_apex_idx
224
+ while end + 1 < len(all_apex_mz) and (all_apex_mz[end + 1] - all_apex_mz[end]) <= gap_threshold:
225
+ end += 1
226
+ while start - 1 >= 0 and (all_apex_mz[start] - all_apex_mz[start - 1]) <= gap_threshold:
227
+ start -= 1
228
+
229
+ apex_mz = all_apex_mz[start : end + 1]
230
+ apex_int = all_apex_int[start : end + 1]
231
+
232
+ logger.debug(
233
+ f'Kept {len(apex_mz)} contiguous apex points '
234
+ f'[{apex_mz[0]:.4f}, {apex_mz[-1]:.4f}] around max at '
235
+ f'{all_apex_mz[max_apex_idx]:.4f} (gap_threshold={gap_threshold:.3f})'
236
+ )
237
+
238
+ if len(apex_mz) < 3:
239
+ logger.debug(f'Too few contiguous points, using all {len(all_apex_mz)} apex points')
240
+ apex_mz = all_apex_mz
241
+ apex_int = all_apex_int
242
+
243
+ # CHECK FOR ALTERNATING INTENSITY PATTERN (same logic as charge detection)
244
+ # At low charge states (z=2), isotope peaks are ~0.5 Da apart with deep
245
+ # valleys; find_peaks picks up both real isotope apexes AND minor peaks
246
+ # in the valleys, creating a high-low-high-low pattern that shifts the
247
+ # smooth envelope centroid. Replace minor peaks' intensities with
248
+ # interpolated values from major peaks to correct the envelope shape
249
+ # while preserving the full m/z range for display.
250
+ if len(apex_int) >= 4:
251
+ intensity_diffs = np.diff(apex_int)
252
+ signs = np.sign(intensity_diffs)
253
+ sign_changes = np.diff(signs)
254
+ alternation_ratio = np.sum(sign_changes != 0) / len(sign_changes) if len(sign_changes) > 0 else 0
255
+
256
+ if alternation_ratio > 0.8:
257
+ even_sum = np.sum(apex_int[0::2])
258
+ odd_sum = np.sum(apex_int[1::2])
259
+ if even_sum >= odd_sum:
260
+ major_idx = np.arange(0, len(apex_int), 2)
261
+ minor_idx = np.arange(1, len(apex_int), 2)
262
+ else:
263
+ major_idx = np.arange(1, len(apex_int), 2)
264
+ minor_idx = np.arange(0, len(apex_int), 2)
265
+
266
+ # Interpolate minor peak intensities from major peaks
267
+ interp_int = np.interp(apex_mz[minor_idx], apex_mz[major_idx], apex_int[major_idx])
268
+ apex_int = apex_int.copy()
269
+ apex_int[minor_idx] = interp_int
270
+ logger.info(
271
+ f'Alternating pattern detected (ratio={alternation_ratio:.2f}): '
272
+ f'interpolated {len(minor_idx)} minor peaks from {len(major_idx)} major peaks'
273
+ )
274
+
275
+ # Create SMOOTH envelope by interpolating apex points + Gaussian smoothing
276
+ # STEP 1: Interpolate apex points to create smooth curve
277
+ # STEP 2: Apply Gaussian smoothing based on instrument resolution
278
+ from scipy.interpolate import UnivariateSpline
279
+ from scipy.ndimage import gaussian_filter1d
280
+
281
+ # Create fine m/z grid
282
+ mz_min = np.min(apex_mz)
283
+ mz_max = np.max(apex_mz)
284
+ num_points = int((mz_max - mz_min) / (fwhm / 100)) + 1
285
+ mz_grid = np.linspace(mz_min, mz_max, num_points)
286
+
287
+ # STEP 1: Interpolate apex points with cubic spline
288
+ if len(apex_mz) >= 4:
289
+ spline = UnivariateSpline(apex_mz, apex_int, s=0, k=3) # cubic, no smoothing
290
+ intensity_interp = spline(mz_grid)
291
+ logger.debug(f'STEP 1: Cubic spline through {len(apex_mz)} apex -> {len(mz_grid)} points')
292
+ else:
293
+ intensity_interp = np.interp(mz_grid, apex_mz, apex_int)
294
+ logger.debug(f'STEP 1: Linear interpolation through {len(apex_mz)} apex -> {len(mz_grid)} points')
295
+
296
+ # STEP 2: Apply STRONGER Gaussian smoothing for better curve fitting
297
+ mz_step = (mz_max - mz_min) / num_points if num_points > 1 else fwhm / 100
298
+ sigma_pixels = (sigma / mz_step) * 15.0 # 15x stronger smoothing for better Gaussian fit
299
+ intensity_grid = gaussian_filter1d(intensity_interp, sigma=sigma_pixels, mode='nearest')
300
+ logger.debug(f'STEP 2: STRONG Gaussian smoothing (sigma={sigma:.6f} m/z x 15 = {sigma_pixels:.2f} pixels)')
301
+
302
+ # Clip negative values (artifacts from edge smoothing)
303
+ intensity_grid = np.maximum(intensity_grid, 0.0)
304
+
305
+ # Normalize to 100
306
+ if np.max(intensity_grid) > 0:
307
+ intensity_grid = (intensity_grid / np.max(intensity_grid)) * 100.0
308
+
309
+ logger.info(f'SUCCESS: Smooth envelope from {len(apex_mz)} apex points')
310
+ logger.debug(f'Envelope: {len(mz_grid)} points, m/z [{np.min(mz_grid):.4f}, {np.max(mz_grid):.4f}]')
311
+
312
+ return mz_grid, intensity_grid
313
+
314
+ except Exception as e:
315
+ logger.exception(f'[generate_experimental_gaussian_envelope] Exception: {str(e)}')
316
+ return None, None
317
+
318
+ def fit_gaussian_to_smooth_envelope(
319
+ self,
320
+ mz_array: Optional[npt.NDArray[np.float64]],
321
+ int_array: Optional[npt.NDArray[np.float64]],
322
+ resolution: int,
323
+ context: str = '',
324
+ ) -> tuple[Optional[float], Optional[float], bool]:
325
+ """
326
+ Fit Gaussian to pre-smoothed isotope envelope to extract X₀ (centroid) and σ.
327
+
328
+ This is used by routes to fit experimental envelopes after they've been
329
+ smoothed by generate_experimental_gaussian_envelope().
330
+
331
+ Approach:
332
+ 1. Find apex points in the data
333
+ 2. Find valley boundaries (left/right) by scanning from center
334
+ 3. Fit Gaussian to ALL data points between valleys
335
+
336
+ Args:
337
+ mz_array: Pre-smoothed m/z values
338
+ int_array: Pre-smoothed intensity values
339
+ resolution: Instrument resolution (for initial sigma estimate)
340
+ context: Optional context string for debug messages
341
+
342
+ Returns:
343
+ (x0, sigma, fit_succeeded): Fitted centroid and width, with success flag
344
+ Falls back to apex values if fit fails (fit_succeeded=False)
345
+ """
346
+ from scipy.signal import find_peaks
347
+
348
+ # Igor Pro-style 4-parameter Gaussian: f(x) = y0 + A × exp(-((x - x₀) / w)²)
349
+ def gaussian(x, y0, A, x0, width):
350
+ return y0 + A * np.exp(-(((x - x0) / width) ** 2))
351
+
352
+ if mz_array is None or int_array is None or len(mz_array) <= 3:
353
+ return None, None, False
354
+
355
+ mz_array = np.array(mz_array)
356
+ int_array = np.array(int_array)
357
+
358
+ try:
359
+ # Step 1: Find apex points to determine valley boundaries
360
+ peaks_idx, _ = find_peaks(int_array, distance=2, prominence=np.max(int_array) * 0.05)
361
+
362
+ if len(peaks_idx) >= 5:
363
+ mz_apex = mz_array[peaks_idx]
364
+ int_apex = int_array[peaks_idx]
365
+
366
+ # Find center (highest apex)
367
+ center_idx = np.argmax(int_apex)
368
+
369
+ # Scan left to find valley
370
+ left_bound_idx = 0
371
+ for i in range(center_idx - 1, 0, -1):
372
+ if i > 0 and int_apex[i] < int_apex[i - 1]:
373
+ if int_apex[i] < int_apex[i + 1] * 0.9:
374
+ left_bound_idx = i
375
+ break
376
+
377
+ # Scan right to find valley
378
+ right_bound_idx = len(int_apex) - 1
379
+ for i in range(center_idx + 1, len(int_apex)):
380
+ if i < len(int_apex) - 1 and int_apex[i] < int_apex[i + 1]:
381
+ if int_apex[i] < int_apex[i - 1] * 0.9:
382
+ right_bound_idx = i
383
+ break
384
+
385
+ # Get m/z boundaries from valleys
386
+ left_mz = mz_apex[left_bound_idx]
387
+ right_mz = mz_apex[right_bound_idx]
388
+
389
+ # Extract ALL data points between valleys
390
+ mask = (mz_array >= left_mz) & (mz_array <= right_mz)
391
+ mz_fit = mz_array[mask]
392
+ int_fit = int_array[mask]
393
+
394
+ if context:
395
+ logger.debug(
396
+ f'[{context}] Valley boundaries: [{left_mz:.4f}, {right_mz:.4f}], fitting {len(mz_fit)} points'
397
+ )
398
+ else:
399
+ # Fallback: use all data
400
+ mz_fit = mz_array
401
+ int_fit = int_array
402
+ if context:
403
+ logger.debug(f'[{context}] Too few apexes ({len(peaks_idx)}), using all {len(mz_fit)} points')
404
+
405
+ if len(mz_fit) < 3:
406
+ mz_fit = mz_array
407
+ int_fit = int_array
408
+
409
+ # Step 2: Fit Igor-style 4-parameter Gaussian to data between valleys
410
+ max_idx = np.argmax(int_fit)
411
+ A_init = int_fit[max_idx]
412
+ x0_init = mz_fit[max_idx]
413
+ fwhm_estimate = x0_init / resolution
414
+ sigma_init = fwhm_estimate / 2.355
415
+ width_init = sigma_init * np.sqrt(2)
416
+ y0_init = np.min(int_fit)
417
+
418
+ popt, pcov = curve_fit(
419
+ gaussian,
420
+ mz_fit,
421
+ int_fit,
422
+ p0=[y0_init, A_init, x0_init, width_init],
423
+ bounds=(
424
+ [-A_init * 0.1, 0, mz_fit[0], 0.001],
425
+ [A_init * 0.5, A_init * 2, mz_fit[-1], fwhm_estimate * 2 * np.sqrt(2)],
426
+ ),
427
+ maxfev=5000,
428
+ )
429
+
430
+ x0 = float(popt[2])
431
+ sigma = float(abs(popt[3]) / np.sqrt(2))
432
+
433
+ if context:
434
+ logger.debug(f'[{context}] Gaussian fit: X0={x0:.4f} m/z, sigma={sigma:.6f} m/z')
435
+
436
+ return x0, sigma, True
437
+
438
+ except Exception as e:
439
+ if context:
440
+ logger.warning(f'[{context}] Gaussian fit failed ({str(e)}), using apex fallback')
441
+ # Fallback: use apex of smooth envelope
442
+ max_idx = np.argmax(int_array)
443
+ x0 = float(mz_array[max_idx])
444
+ fwhm = x0 / resolution
445
+ sigma = float(fwhm / 2.355)
446
+
447
+ return x0, sigma, False
448
+
449
+ def detect_peak_asymmetry_visual(
450
+ self, mz_array: npt.NDArray[np.float64], int_array: npt.NDArray[np.float64], threshold_ratio: float = 0.3
451
+ ) -> tuple[bool, int, str]:
452
+ """
453
+ Detect peak asymmetry using visual characteristics:
454
+ - Count local maxima (multiple bumps = asymmetric)
455
+ - Check for shoulders (secondary peaks)
456
+ - Measure envelope smoothness
457
+
458
+ Returns: (is_asymmetric, num_maxima, details)
459
+ """
460
+ try:
461
+ if len(mz_array) < 5:
462
+ return False, 1, 'Too few points'
463
+
464
+ mz_array = np.array(mz_array)
465
+ int_array = np.array(int_array)
466
+
467
+ # Normalize intensity
468
+ max_int = np.max(int_array)
469
+ if max_int == 0:
470
+ return False, 1, 'Zero intensity'
471
+
472
+ int_norm = int_array / max_int
473
+
474
+ # Find local maxima (peaks)
475
+ from scipy.signal import find_peaks
476
+
477
+ # Detect peaks with minimum height (to avoid noise)
478
+ # Prominence helps identify significant peaks vs noise
479
+ peaks, properties = find_peaks(
480
+ int_norm,
481
+ height=threshold_ratio, # At least 30% of max height
482
+ prominence=0.1, # Must be prominent enough
483
+ distance=3, # Separated by at least 3 points
484
+ )
485
+
486
+ num_maxima = len(peaks)
487
+
488
+ # Determine if asymmetric based on number of significant maxima
489
+ is_asymmetric = num_maxima > 1
490
+
491
+ details = f'{num_maxima} local maxima detected'
492
+ if num_maxima > 1:
493
+ peak_positions = [f'{mz_array[p]:.2f}' for p in peaks]
494
+ details += f' at m/z: {", ".join(peak_positions)}'
495
+
496
+ logger.debug(f'[Visual asymmetry detection] {details} -> {"ASYMMETRIC" if is_asymmetric else "SYMMETRIC"}')
497
+
498
+ return is_asymmetric, num_maxima, details
499
+
500
+ except Exception as e:
501
+ logger.error(f'[detect_peak_asymmetry_visual] Error: {str(e)}')
502
+ return False, 1, f'Error: {str(e)}'
503
+
504
+ def calculate_peak_skewness(
505
+ self, mz_array: npt.NDArray[np.float64], int_array: npt.NDArray[np.float64]
506
+ ) -> Optional[float]:
507
+ """
508
+ Calculate peak skewness (asymmetry measure).
509
+ Skewness = 0: perfectly symmetric
510
+ Skewness > 0: right-tailed (tailing to higher m/z)
511
+ Skewness < 0: left-tailed (tailing to lower m/z)
512
+
513
+ Returns: skewness value
514
+ """
515
+ try:
516
+ mz_array = np.array(mz_array, dtype=float)
517
+ int_array = np.array(int_array, dtype=float)
518
+
519
+ if len(mz_array) < 3 or np.sum(int_array) == 0:
520
+ return None
521
+
522
+ # Calculate mean (weighted by intensity)
523
+ mean = np.sum(mz_array * int_array) / np.sum(int_array)
524
+
525
+ # Calculate standard deviation
526
+ variance = np.sum(int_array * (mz_array - mean) ** 2) / np.sum(int_array)
527
+ std_dev = np.sqrt(variance)
528
+
529
+ if std_dev == 0:
530
+ return 0.0
531
+
532
+ # Calculate skewness: (mean - mode) / std_dev
533
+ # Approximate mode as the m/z with maximum intensity
534
+ mode_idx = np.argmax(int_array)
535
+ mode = mz_array[mode_idx]
536
+
537
+ skewness = (mean - mode) / std_dev
538
+
539
+ return float(skewness)
540
+
541
+ except Exception as e:
542
+ logger.error(f'[calculate_peak_skewness] Exception: {str(e)}')
543
+ return None
544
+
545
+ def weighted_average_centroid(
546
+ self, mz_array: npt.NDArray[np.float64], int_array: npt.NDArray[np.float64]
547
+ ) -> tuple[Optional[float], Optional[float]]:
548
+ """
549
+ Calculate centroid using weighted average method.
550
+ This is used for general centroid calculations.
551
+ Returns x₀ = Σ(m/z × intensity) / Σ(intensity) and σ (weighted std dev)
552
+ """
553
+ try:
554
+ if len(mz_array) == 0 or len(int_array) == 0:
555
+ logger.warning('[weighted_average_centroid] Empty arrays')
556
+ return None, None
557
+
558
+ # Convert to numpy arrays
559
+ mz_array = np.array(mz_array, dtype=float)
560
+ int_array = np.array(int_array, dtype=float)
561
+
562
+ total_intensity = np.sum(int_array)
563
+ if total_intensity == 0 or np.isnan(total_intensity) or np.isinf(total_intensity):
564
+ logger.warning(f'[weighted_average_centroid] Invalid total intensity: {total_intensity}')
565
+ return None, None
566
+
567
+ # Weighted average: x₀ = Σ(m/z × intensity) / Σ(intensity)
568
+ x0 = np.sum(mz_array * int_array) / total_intensity
569
+
570
+ # Weighted standard deviation: σ = sqrt(Σ(intensity × (m/z - x₀)²) / Σ(intensity))
571
+ sigma = np.sqrt(np.sum(int_array * (mz_array - x0) ** 2) / total_intensity)
572
+
573
+ if np.isnan(x0) or np.isinf(x0):
574
+ return None, None
575
+
576
+ logger.debug(f'[weighted_average_centroid] x0={x0:.4f}, sigma={sigma:.4f}')
577
+ return float(x0), float(sigma)
578
+
579
+ except Exception as e:
580
+ logger.error(f'[weighted_average_centroid] Exception: {str(e)}')
581
+ return None, None
582
+
583
+ def gaussian_fit_centroid(
584
+ self, mz_array: npt.NDArray[np.float64], int_array: npt.NDArray[np.float64], return_quality: bool = False
585
+ ) -> tuple:
586
+ """
587
+ Fit Gaussian curve to isotope envelope: f(x) = A × exp(-(x - x₀)² / (2σ²))
588
+ This is the STANDARD method used for composition determination (X₀ error calculation).
589
+
590
+ Parameters:
591
+ - return_quality: If True, also return R² goodness-of-fit metric
592
+
593
+ Returns:
594
+ - (x0, sigma, x0_error) if return_quality=False
595
+ - (x0, sigma, x0_error, r_squared) if return_quality=True
596
+ - x0_error is the standard error of the fitted x₀ parameter from covariance matrix
597
+ """
598
+ try:
599
+ if len(mz_array) == 0 or len(int_array) == 0:
600
+ logger.warning(
601
+ f'[gaussian_fit_centroid] Empty arrays: mz length={len(mz_array)}, int length={len(int_array)}'
602
+ )
603
+ if return_quality:
604
+ return None, None, None, None
605
+ else:
606
+ return None, None, None
607
+
608
+ # Convert to numpy arrays
609
+ mz_array = np.array(mz_array, dtype=float)
610
+ int_array = np.array(int_array, dtype=float)
611
+
612
+ if len(mz_array) < 3:
613
+ logger.warning('[gaussian_fit_centroid] Need at least 3 points for fitting')
614
+ if return_quality:
615
+ return None, None, None, None
616
+ else:
617
+ return None, None, None
618
+
619
+ total_intensity = np.sum(int_array)
620
+ if total_intensity == 0 or np.isnan(total_intensity) or np.isinf(total_intensity):
621
+ logger.warning(f'[gaussian_fit_centroid] Invalid total intensity: {total_intensity}')
622
+ if return_quality:
623
+ return None, None, None, None
624
+ else:
625
+ return None, None, None
626
+
627
+ # NEW APPROACH: Find apex envelope, then find valleys in envelope, then fit
628
+ # Step 1: Find apex points of individual isotope peaks
629
+ from scipy.signal import find_peaks
630
+
631
+ # Find local maxima (apex of each isotope peak)
632
+ peaks_idx, _ = find_peaks(int_array, distance=2, prominence=np.max(int_array) * 0.05)
633
+
634
+ if len(peaks_idx) >= 5:
635
+ # Extract apex points (envelope)
636
+ mz_apex = mz_array[peaks_idx]
637
+ int_apex = int_array[peaks_idx]
638
+ logger.debug(f'[gaussian_fit_centroid] Found {len(peaks_idx)} isotope peak apexes')
639
+
640
+ # Step 2: Find the highest apex (center of envelope)
641
+ center_idx = np.argmax(int_apex)
642
+
643
+ # Step 3: Find valleys in the envelope (left and right of center)
644
+ # A valley must be both a local dip (relative criterion) AND
645
+ # below 40% of center intensity (absolute criterion) to avoid
646
+ # triggering on noise fluctuations in broad complex envelopes.
647
+ center_intensity = int_apex[center_idx]
648
+ valley_threshold = center_intensity * 0.4
649
+
650
+ # Scan left from center to find minimum
651
+ left_bound_idx = 0
652
+ for i in range(center_idx - 1, 0, -1):
653
+ if i > 0 and int_apex[i] < int_apex[i - 1]:
654
+ if int_apex[i] < int_apex[i + 1] * 0.9 and int_apex[i] < valley_threshold:
655
+ left_bound_idx = i
656
+ break
657
+
658
+ # Scan right from center to find minimum
659
+ right_bound_idx = len(int_apex) - 1
660
+ for i in range(center_idx + 1, len(int_apex)):
661
+ if i < len(int_apex) - 1 and int_apex[i] < int_apex[i + 1]:
662
+ if int_apex[i] < int_apex[i - 1] * 0.9 and int_apex[i] < valley_threshold:
663
+ right_bound_idx = i
664
+ break
665
+
666
+ # Step 4: Extract apex points BETWEEN envelope valleys
667
+ mz_fit = mz_apex[left_bound_idx : right_bound_idx + 1]
668
+ int_fit = int_apex[left_bound_idx : right_bound_idx + 1]
669
+
670
+ logger.debug(
671
+ f'[gaussian_fit_centroid] Envelope valleys: left={left_bound_idx}, center={center_idx}, right={right_bound_idx}'
672
+ )
673
+ logger.debug(f'[gaussian_fit_centroid] Fitting to {len(mz_fit)} apex points between envelope valleys')
674
+ else:
675
+ # Fallback: if too few peaks, use all apex points or top 70%
676
+ if len(peaks_idx) >= 3:
677
+ mz_fit = mz_array[peaks_idx]
678
+ int_fit = int_array[peaks_idx]
679
+ logger.warning(f'[gaussian_fit_centroid] Only {len(peaks_idx)} apexes, using all')
680
+ else:
681
+ logger.warning('[gaussian_fit_centroid] Too few apexes, using top 70%')
682
+ max_intensity = np.max(int_array)
683
+ threshold = max_intensity * 0.70
684
+ high_intensity_mask = int_array >= threshold
685
+ mz_fit = mz_array[high_intensity_mask]
686
+ int_fit = int_array[high_intensity_mask]
687
+
688
+ if len(mz_fit) < 3:
689
+ logger.error('[gaussian_fit_centroid] Too few points, using all data')
690
+ mz_fit = mz_array
691
+ int_fit = int_array
692
+
693
+ # Initial guesses for Gaussian parameters
694
+ # Amplitude: maximum intensity
695
+ A_guess = np.max(int_fit)
696
+
697
+ # Center: m/z of the maximum intensity point (apex)
698
+ max_idx = np.argmax(int_fit)
699
+ x0_guess = mz_fit[max_idx]
700
+
701
+ # Width: estimate from data range
702
+ mz_min_fit = np.min(mz_fit)
703
+ mz_max_fit = np.max(mz_fit)
704
+ mz_range_fit = mz_max_fit - mz_min_fit
705
+ sigma_guess = mz_range_fit / 4.0 # Narrower estimate since we're fitting top only
706
+
707
+ # Ensure reasonable initial guesses
708
+ if sigma_guess < 0.01:
709
+ sigma_guess = 0.5
710
+
711
+ # Overall data range for bounds
712
+ mz_min_all = np.min(mz_array)
713
+ mz_max_all = np.max(mz_array)
714
+
715
+ # Baseline guess: minimum intensity in fitting region
716
+ max_int_fit = np.max(int_fit)
717
+ y0_guess = min(np.min(int_fit), max_int_fit * 0.4)
718
+
719
+ logger.debug(
720
+ f'[gaussian_fit_centroid] Initial guesses: x0={x0_guess:.4f} (apex), sigma={sigma_guess:.4f}, A={A_guess:.2e}, y0={y0_guess:.2e}'
721
+ )
722
+
723
+ # Igor Pro-style 4-parameter Gaussian: f(x) = y0 + A × exp(-((x - x₀) / w)²)
724
+ # where w = sqrt(2) × σ, so exponent = -(x - x₀)² / (2σ²)
725
+ def gaussian(x, y0, A, x0, width):
726
+ return y0 + A * np.exp(-(((x - x0) / width) ** 2))
727
+
728
+ # Fit Gaussian curve to HIGH-INTENSITY data only
729
+ try:
730
+ from scipy.optimize import curve_fit
731
+
732
+ width_guess = sigma_guess * np.sqrt(2)
733
+
734
+ # Allow x0 to vary within the full valley boundaries
735
+ bounds = (
736
+ [-max_int_fit * 0.1, 0, mz_min_all, 0.01], # Lower bounds: [y0_min, A_min, x0_min, width_min]
737
+ [max_int_fit * 0.5, np.inf, mz_max_all, (mz_max_all - mz_min_all) * 2], # Upper bounds
738
+ )
739
+
740
+ logger.debug(f'[gaussian_fit_centroid] x0 bounds: [{mz_min_all:.4f}, {mz_max_all:.4f}]')
741
+
742
+ popt, pcov = curve_fit(
743
+ gaussian,
744
+ mz_fit, # Fit to high-intensity points only
745
+ int_fit,
746
+ p0=[y0_guess, A_guess, x0_guess, width_guess],
747
+ bounds=bounds,
748
+ maxfev=10000,
749
+ ftol=1e-10, # Function tolerance for convergence (more precise)
750
+ xtol=1e-10, # Parameter tolerance for convergence (more precise)
751
+ )
752
+
753
+ y0_fit, A_fit, x0_fit, width_fit = popt
754
+ sigma_fit = width_fit / np.sqrt(2)
755
+
756
+ # Calculate standard errors from covariance matrix
757
+ # pcov diagonal gives variance of parameters, sqrt gives standard error
758
+ perr = np.sqrt(np.diag(pcov))
759
+ y0_err, A_err, x0_err, width_err = perr
760
+
761
+ # Validate fitted parameters
762
+ if np.isnan(x0_fit) or np.isinf(x0_fit) or np.isnan(sigma_fit) or np.isinf(sigma_fit):
763
+ raise ValueError('Fit returned invalid parameters')
764
+
765
+ # Calculate R² (coefficient of determination) if requested
766
+ if return_quality:
767
+ # Predicted values from fitted Gaussian
768
+ y_pred = gaussian(mz_array, y0_fit, A_fit, x0_fit, width_fit)
769
+
770
+ # Calculate R²
771
+ ss_res = np.sum((int_array - y_pred) ** 2) # Residual sum of squares
772
+ ss_tot = np.sum((int_array - np.mean(int_array)) ** 2) # Total sum of squares
773
+ r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0.0
774
+
775
+ logger.debug(
776
+ f'[gaussian_fit_centroid] Gaussian fit: x₀={x0_fit:.4f}±{x0_err:.4f}, σ={sigma_fit:.4f}, y0={y0_fit:.2e}, R²={r_squared:.4f}'
777
+ )
778
+ return float(x0_fit), float(sigma_fit), float(x0_err), float(r_squared)
779
+ else:
780
+ logger.debug(
781
+ f'[gaussian_fit_centroid] Gaussian fit: x₀={x0_fit:.4f}±{x0_err:.4f}, σ={sigma_fit:.4f}, y0={y0_fit:.2e}'
782
+ )
783
+ return float(x0_fit), float(sigma_fit), float(x0_err)
784
+
785
+ except Exception as fit_error:
786
+ # If Gaussian fit fails, fall back to weighted average
787
+ logger.warning(
788
+ f'[gaussian_fit_centroid] Gaussian fit failed ({fit_error}), using weighted average fallback'
789
+ )
790
+ x0_fallback = x0_guess
791
+ sigma_fallback = sigma_guess
792
+
793
+ if np.isnan(x0_fallback) or np.isinf(x0_fallback):
794
+ if return_quality:
795
+ return None, None, None, None
796
+ else:
797
+ return None, None, None
798
+
799
+ if return_quality:
800
+ return (
801
+ float(x0_fallback),
802
+ float(sigma_fallback),
803
+ None,
804
+ 0.0,
805
+ ) # No fitting error, R² = 0 indicates fit failed
806
+ else:
807
+ return float(x0_fallback), float(sigma_fallback), None # No fitting error available
808
+
809
+ except Exception as e:
810
+ logger.exception(f'[gaussian_fit_centroid] Exception: {str(e)}')
811
+ if return_quality:
812
+ return None, None, None, None
813
+ else:
814
+ return None, None, None
core/isotope.py ADDED
@@ -0,0 +1,277 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import logging
4
+ import os
5
+ import sys
6
+ from typing import Any
7
+
8
+ import numpy as np
9
+
10
+ current_dir = os.path.dirname(os.path.abspath(__file__))
11
+ sys.path.insert(0, os.path.join(current_dir, '..', 'lib'))
12
+
13
+ from pythoms.molecule import IPMolecule
14
+
15
+ try:
16
+ import IsoSpecPy as isospec
17
+ ISOSPEC_AVAILABLE = True
18
+ except ImportError:
19
+ ISOSPEC_AVAILABLE = False
20
+
21
+ ISOTOPE_LIBRARY = 'isospec' if ISOSPEC_AVAILABLE else 'pythoms'
22
+
23
+ _isotope_pattern_cache: dict[tuple[str, int, int], dict[str, Any]] = {}
24
+ _ISOTOPE_CACHE_MAX_SIZE = 1000
25
+
26
+ logger = logging.getLogger(__name__)
27
+
28
+
29
+ class IsotopeMixin:
30
+ """Mixin for isotope pattern generation (IsoSpecPy and PythoMS backends)."""
31
+
32
+ def generate_isotope_pattern(self, formula: str, charge: int = 1, resolution: int = 20000) -> dict:
33
+ """
34
+ Generate isotope pattern for a given formula.
35
+ Dispatches to either IsoSpecPy (faster) or PythoMS based on ISOTOPE_LIBRARY setting.
36
+ Returns both bar pattern and Gaussian pattern.
37
+
38
+ Uses global cache for speed optimization.
39
+
40
+ Parameters:
41
+ formula: Chemical formula string
42
+ charge: Charge state
43
+ resolution: MS resolution (default 20000 is fallback when webapp cannot parse from uploaded data)
44
+ """
45
+ global _isotope_pattern_cache, _ISOTOPE_CACHE_MAX_SIZE, ISOTOPE_LIBRARY
46
+
47
+ # Check cache first
48
+ cache_key = (formula, charge, resolution)
49
+ if cache_key in _isotope_pattern_cache:
50
+ logger.debug(f'[generate_isotope_pattern] CACHE HIT for {formula[:30]}... (z={charge})')
51
+ return _isotope_pattern_cache[cache_key]
52
+
53
+ logger.debug(f'[generate_isotope_pattern] CACHE MISS for {formula[:30]}... (z={charge}) - computing...')
54
+ # Dispatch to appropriate library
55
+ if ISOTOPE_LIBRARY == 'isospec' and ISOSPEC_AVAILABLE:
56
+ result = self._generate_isotope_pattern_isospec(formula, charge, resolution)
57
+ else:
58
+ result = self._generate_isotope_pattern_pythoms(formula, charge, resolution)
59
+
60
+ # Cache the result (with size limit) if successful
61
+ if 'error' not in result:
62
+ if len(_isotope_pattern_cache) >= _ISOTOPE_CACHE_MAX_SIZE:
63
+ # Remove oldest entry (first key)
64
+ oldest_key = next(iter(_isotope_pattern_cache))
65
+ del _isotope_pattern_cache[oldest_key]
66
+ _isotope_pattern_cache[cache_key] = result
67
+
68
+ return result
69
+
70
+ def _consolidate_formula(self, formula: str) -> str:
71
+ """
72
+ Consolidate a formula with duplicate elements into standard form.
73
+ E.g., 'C304H368N128O184P30Ag28N2H8' -> 'C304H376N130O184P30Ag28'
74
+
75
+ IsoSpecPy requires each element to appear only once.
76
+ """
77
+ import re
78
+
79
+ # Parse formula: find all element-count pairs
80
+ # Matches element symbols (1-2 letters, first uppercase) followed by optional count
81
+ pattern = r'([A-Z][a-z]?)(\d*)'
82
+ matches = re.findall(pattern, formula)
83
+
84
+ # Consolidate counts for each element
85
+ element_counts: dict[str, int] = {}
86
+ for element, count in matches:
87
+ if element: # Skip empty matches
88
+ count = int(count) if count else 1
89
+ element_counts[element] = element_counts.get(element, 0) + count
90
+
91
+ # Rebuild formula in a standard order (C, H, N, O, P, S, then others alphabetically)
92
+ priority_order = ['C', 'H', 'N', 'O', 'P', 'S']
93
+ result = []
94
+
95
+ # Add priority elements first
96
+ for elem in priority_order:
97
+ if elem in element_counts:
98
+ count = element_counts.pop(elem)
99
+ result.append(f'{elem}{count}' if count > 1 else elem)
100
+
101
+ # Add remaining elements alphabetically
102
+ for elem in sorted(element_counts.keys()):
103
+ count = element_counts[elem]
104
+ result.append(f'{elem}{count}' if count > 1 else elem)
105
+
106
+ return ''.join(result)
107
+
108
+ def _generate_isotope_pattern_isospec(self, formula: str, charge: int = 1, resolution: int = 20000) -> dict:
109
+ """
110
+ Generate isotope pattern using IsoSpecPy (faster than PythoMS for large molecules).
111
+ """
112
+ try:
113
+ # Consolidate formula to handle duplicate elements (e.g., from adducts)
114
+ # IsoSpecPy requires each element to appear only once
115
+ consolidated_formula = self._consolidate_formula(formula)
116
+
117
+ # prob_to_cover=0.9999 captures 99.99% of the isotope distribution
118
+ iso_result = isospec.IsoTotalProb(formula=consolidated_formula, prob_to_cover=0.9999)
119
+
120
+ # IsoSpecPy returns CFFI objects - must convert to list first
121
+ masses = np.array(list(iso_result.masses))
122
+ probs = np.array(list(iso_result.probs))
123
+
124
+ if len(masses) == 0:
125
+ return {'error': 'IsoSpecPy returned empty pattern'}
126
+
127
+ # Sort by mass first
128
+ sort_idx = np.argsort(masses)
129
+ masses = masses[sort_idx]
130
+ probs = probs[sort_idx]
131
+
132
+ # Get monoisotopic mass (first peak after sorting, before any filtering)
133
+ monoisotopic_mass = masses[0]
134
+
135
+ # Calculate molecular weight (weighted average of ALL peaks)
136
+ molecular_weight = np.average(masses, weights=probs)
137
+
138
+ # Convert to m/z (apply charge)
139
+ # The formula passed is already the ION formula (protons already removed)
140
+ # So we just divide by charge, same as PythoMS does
141
+ mz_values = masses / abs(charge)
142
+
143
+ # Bin peaks FIRST to match PythoMS behavior (combine peaks at similar m/z)
144
+ # IsoSpecPy returns fine-grained peaks, PythoMS aggregates by nominal mass
145
+ # Use 0.2 Da bins to match typical isotope spacing
146
+ bin_width = 0.2 / abs(charge)
147
+ min_mz = mz_values.min()
148
+ max_mz = mz_values.max()
149
+ bins = np.arange(min_mz - bin_width / 2, max_mz + bin_width, bin_width)
150
+
151
+ # Digitize: assign each peak to a bin
152
+ bin_indices = np.digitize(mz_values, bins)
153
+
154
+ # Aggregate peaks in each bin
155
+ binned_mz = []
156
+ binned_int = []
157
+ for i in range(1, len(bins)):
158
+ mask = bin_indices == i
159
+ if np.any(mask):
160
+ # Weighted average for m/z, sum for intensity
161
+ bin_probs = probs[mask]
162
+ bin_mzs = mz_values[mask]
163
+ binned_mz.append(np.average(bin_mzs, weights=bin_probs))
164
+ binned_int.append(np.sum(bin_probs))
165
+
166
+ if not binned_mz:
167
+ return {'error': 'IsoSpecPy: no peaks after binning'}
168
+
169
+ mz_values = np.array(binned_mz)
170
+ probs = np.array(binned_int)
171
+
172
+ # Normalize AFTER binning to max = 1.0 (like PythoMS)
173
+ probs = probs / np.max(probs)
174
+
175
+ # Filter out low intensity peaks (threshold=0.01 like PythoMS)
176
+ threshold = 0.01
177
+ mask = probs >= threshold
178
+ mz_values = mz_values[mask]
179
+ probs = probs[mask]
180
+
181
+ if len(mz_values) == 0:
182
+ return {'error': 'IsoSpecPy: all peaks below threshold after filtering'}
183
+
184
+ # Create bar isotope pattern in PythoMS format
185
+ barip = [mz_values.tolist(), probs.tolist()]
186
+
187
+ # Calculate FWHM for Gaussian smoothing
188
+ if len(mz_values) > 0:
189
+ theoretical_mz = mz_values[0]
190
+ else:
191
+ theoretical_mz = monoisotopic_mass / abs(charge)
192
+
193
+ fwhm = theoretical_mz / resolution
194
+
195
+ # Generate Gaussian pattern using the same smooth function
196
+ gaussian_pattern = self.smooth_gaussian_pattern(barip, fwhm, num_points_per_fwhm=100)
197
+
198
+ # Sort Gaussian pattern by m/z
199
+ if gaussian_pattern and len(gaussian_pattern[0]) > 0:
200
+ gaussian_mz = np.array(gaussian_pattern[0])
201
+ gaussian_int = np.array(gaussian_pattern[1])
202
+ sort_idx = np.argsort(gaussian_mz)
203
+ gaussian_mz_sorted = gaussian_mz[sort_idx].tolist()
204
+ gaussian_int_sorted = gaussian_int[sort_idx].tolist()
205
+ else:
206
+ gaussian_mz_sorted = []
207
+ gaussian_int_sorted = []
208
+
209
+ return {
210
+ 'mz': barip[0],
211
+ 'intensity': barip[1],
212
+ 'gaussian_mz': gaussian_mz_sorted,
213
+ 'gaussian_intensity': gaussian_int_sorted,
214
+ 'monoisotopic_mass': monoisotopic_mass,
215
+ 'molecular_weight': molecular_weight,
216
+ }
217
+ except Exception as e:
218
+ # Fall back to PythoMS if IsoSpecPy fails
219
+ logger.warning(f'IsoSpecPy failed for {formula}: {e}, falling back to PythoMS')
220
+ return self._generate_isotope_pattern_pythoms(formula, charge, resolution)
221
+
222
+ def _generate_isotope_pattern_pythoms(self, formula: str, charge: int = 1, resolution: int = 20000) -> dict:
223
+ """
224
+ Generate isotope pattern using PythoMS (original implementation).
225
+ """
226
+ try:
227
+ mol = IPMolecule(
228
+ formula,
229
+ charge=charge,
230
+ resolution=resolution,
231
+ verbose=False,
232
+ ipmethod='hybrid',
233
+ dropmethod='threshold',
234
+ threshold=0.01,
235
+ )
236
+
237
+ # Get bar isotope pattern
238
+ barip = mol.bar_isotope_pattern
239
+
240
+ # Calculate theoretical m/z for FWHM calculation
241
+ # Use the first m/z value from bar pattern (monoisotopic peak)
242
+ if len(barip[0]) > 0:
243
+ theoretical_mz = barip[0][0] # First m/z value = monoisotopic peak
244
+ else:
245
+ # Fallback to old method if bar pattern is empty
246
+ theoretical_mass = mol.monoisotopic_mass
247
+ theoretical_mz = (theoretical_mass - charge * self.m_p) / charge
248
+
249
+ # Generate Gaussian pattern using custom smooth function
250
+ fwhm = theoretical_mz / resolution
251
+
252
+ # Use custom smooth Gaussian generation instead of PythoMS version
253
+ gaussian_pattern = self.smooth_gaussian_pattern(barip, fwhm, num_points_per_fwhm=100)
254
+
255
+ # Sort Gaussian pattern by m/z to prevent zigzag plotting
256
+ if gaussian_pattern and len(gaussian_pattern[0]) > 0:
257
+ gaussian_mz = np.array(gaussian_pattern[0])
258
+ gaussian_int = np.array(gaussian_pattern[1])
259
+
260
+ # Sort by m/z
261
+ sort_idx = np.argsort(gaussian_mz)
262
+ gaussian_mz_sorted = gaussian_mz[sort_idx].tolist()
263
+ gaussian_int_sorted = gaussian_int[sort_idx].tolist()
264
+ else:
265
+ gaussian_mz_sorted = []
266
+ gaussian_int_sorted = []
267
+
268
+ return {
269
+ 'mz': barip[0],
270
+ 'intensity': barip[1],
271
+ 'gaussian_mz': gaussian_mz_sorted,
272
+ 'gaussian_intensity': gaussian_int_sorted,
273
+ 'monoisotopic_mass': mol.monoisotopic_mass,
274
+ 'molecular_weight': mol.molecular_weight,
275
+ }
276
+ except Exception as e:
277
+ return {'error': str(e)}
core/scoring.py ADDED
@@ -0,0 +1,250 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import logging
4
+ from typing import Optional
5
+
6
+ import numpy as np
7
+ import numpy.typing as npt
8
+
9
+ logger = logging.getLogger(__name__)
10
+
11
+
12
+ class ScoringMixin:
13
+ """Mixin for isotope pattern matching and similarity scoring."""
14
+
15
+ def calculate_pattern_similarity(
16
+ self,
17
+ theo_mz: npt.NDArray[np.float64],
18
+ theo_int: npt.NDArray[np.float64],
19
+ exp_mz: npt.NDArray[np.float64],
20
+ exp_int: npt.NDArray[np.float64],
21
+ window: float = 3.0,
22
+ ) -> float:
23
+ """Mean of cosine similarity and Pearson correlation between matched theo/exp isotope peaks."""
24
+ try:
25
+ if len(theo_mz) == 0 or len(exp_mz) == 0:
26
+ return 0.0
27
+
28
+ from scipy.signal import find_peaks
29
+
30
+ theo_mz = np.array(theo_mz, dtype=np.float64)
31
+ theo_int = np.array(theo_int, dtype=np.float64)
32
+ exp_mz = np.array(exp_mz, dtype=np.float64)
33
+ exp_int = np.array(exp_int, dtype=np.float64)
34
+
35
+ # Filter to significant sticks (>5% of max)
36
+ sig_mask = theo_int > np.max(theo_int) * 0.05
37
+ theo_mz = theo_mz[sig_mask]
38
+ theo_int = theo_int[sig_mask]
39
+ if len(theo_mz) < 2:
40
+ return 0.0
41
+
42
+ # Find experimental peak apexes
43
+ spacing = np.median(np.diff(theo_mz))
44
+ min_distance = int(spacing * 0.4 / np.median(np.diff(exp_mz))) if len(exp_mz) > 1 else 2
45
+ peaks_idx, _ = find_peaks(exp_int, distance=max(2, min_distance), prominence=np.max(exp_int) * 0.02)
46
+
47
+ if len(peaks_idx) < 2:
48
+ apex_mz = exp_mz
49
+ apex_int = exp_int
50
+ else:
51
+ apex_mz = exp_mz[peaks_idx]
52
+ apex_int = exp_int[peaks_idx]
53
+
54
+ # Match each stick to nearest apex within half-spacing tolerance
55
+ match_tol = spacing * 0.5
56
+ paired_theo = []
57
+ paired_exp = []
58
+
59
+ for i in range(len(theo_mz)):
60
+ diffs = np.abs(apex_mz - theo_mz[i])
61
+ nearest_idx = np.argmin(diffs)
62
+ if diffs[nearest_idx] <= match_tol:
63
+ paired_theo.append(theo_int[i])
64
+ paired_exp.append(apex_int[nearest_idx])
65
+ else:
66
+ paired_theo.append(theo_int[i])
67
+ paired_exp.append(0.0)
68
+
69
+ if len(paired_theo) < 2:
70
+ return 0.0
71
+
72
+ paired_theo = np.array(paired_theo)
73
+ paired_exp = np.array(paired_exp)
74
+
75
+ # Normalize to max = 1
76
+ paired_theo = paired_theo / (np.max(paired_theo) + 1e-10)
77
+ paired_exp = paired_exp / (np.max(paired_exp) + 1e-10)
78
+
79
+ # Cosine similarity
80
+ cosine_sim = np.dot(paired_theo, paired_exp) / (
81
+ np.linalg.norm(paired_theo) * np.linalg.norm(paired_exp) + 1e-10
82
+ )
83
+ cosine_sim = max(0.0, min(1.0, cosine_sim))
84
+
85
+ # Pearson correlation
86
+ if np.std(paired_theo) > 0 and np.std(paired_exp) > 0:
87
+ correlation = np.corrcoef(paired_theo, paired_exp)[0, 1]
88
+ correlation = max(0.0, min(1.0, correlation))
89
+ else:
90
+ correlation = 0.0
91
+
92
+ return float((cosine_sim + correlation) / 2.0)
93
+
94
+ except Exception as e:
95
+ logger.exception(f'[calculate_pattern_similarity] Exception: {str(e)}')
96
+ return 0.0
97
+
98
+ def calculate_multi_parameter_fit_score(
99
+ self,
100
+ theo_mz: npt.NDArray[np.float64],
101
+ theo_int: npt.NDArray[np.float64],
102
+ exp_mz: npt.NDArray[np.float64],
103
+ exp_int: npt.NDArray[np.float64],
104
+ theo_x0: Optional[float],
105
+ theo_sigma: Optional[float],
106
+ exp_x0: Optional[float],
107
+ exp_sigma: Optional[float],
108
+ ) -> tuple[float, dict]:
109
+ """
110
+ Calculate comprehensive fit score combining multiple parameters:
111
+ 1. X₀ error (centroid position)
112
+ 2. σ ratio (width matching)
113
+ 3. R² (curve overlap quality)
114
+
115
+ Returns a composite score (lower is better) and individual metrics
116
+ """
117
+ try:
118
+ if theo_x0 is None or exp_x0 is None or theo_sigma is None or exp_sigma is None:
119
+ return 999.0, {'x0_error': 999.0, 'sigma_ratio': None, 'r_squared': None}
120
+
121
+ # 1. X₀ error (absolute difference in centroid positions)
122
+ x0_error = abs(exp_x0 - theo_x0)
123
+
124
+ # 2. σ ratio (how well the widths match)
125
+ # Ratio close to 1.0 means good width match
126
+ sigma_ratio = theo_sigma / exp_sigma if exp_sigma > 0 else None
127
+ sigma_deviation = abs(1.0 - sigma_ratio) if sigma_ratio else 999.0
128
+
129
+ # 3. R² (coefficient of determination - curve overlap quality)
130
+ # Need to align theoretical and experimental on same m/z grid
131
+ try:
132
+ # Find overlapping m/z range
133
+ mz_min = max(np.min(theo_mz), np.min(exp_mz))
134
+ mz_max = min(np.max(theo_mz), np.max(exp_mz))
135
+
136
+ if mz_max <= mz_min:
137
+ r_squared = 0.0
138
+ else:
139
+ # Create common m/z grid for comparison
140
+ mz_grid = np.linspace(mz_min, mz_max, 200)
141
+
142
+ # Interpolate both patterns onto common grid
143
+ theo_interp = np.interp(mz_grid, theo_mz, theo_int, left=0, right=0)
144
+ exp_interp = np.interp(mz_grid, exp_mz, exp_int, left=0, right=0)
145
+
146
+ # Normalize both to max=1 for fair comparison
147
+ theo_norm = theo_interp / np.max(theo_interp) if np.max(theo_interp) > 0 else theo_interp
148
+ exp_norm = exp_interp / np.max(exp_interp) if np.max(exp_interp) > 0 else exp_interp
149
+
150
+ # Calculate R² (coefficient of determination)
151
+ ss_res = np.sum((exp_norm - theo_norm) ** 2) # Residual sum of squares
152
+ ss_tot = np.sum((exp_norm - np.mean(exp_norm)) ** 2) # Total sum of squares
153
+ r_squared = 1.0 - (ss_res / ss_tot) if ss_tot > 0 else 0.0
154
+ r_squared = max(0.0, min(1.0, r_squared)) # Clamp to [0, 1]
155
+
156
+ except Exception as e:
157
+ logger.error(f'[R-squared calculation failed]: {str(e)}')
158
+ r_squared = 0.0
159
+
160
+ # Composite score (weighted combination - lower is better)
161
+ # Weight factors - adjust these based on importance
162
+ w_x0 = 10.0 # X₀ error weight (m/z units)
163
+ w_sigma = 5.0 # σ deviation weight
164
+ w_r2 = 20.0 # R² weight (inverted since higher R² is better)
165
+
166
+ composite_score = (
167
+ w_x0 * x0_error # Centroid position error
168
+ + w_sigma * sigma_deviation # Width mismatch
169
+ + w_r2 * (1.0 - r_squared) # Shape overlap quality (inverted)
170
+ )
171
+
172
+ metrics = {
173
+ 'x0_error': float(x0_error),
174
+ 'sigma_ratio': float(sigma_ratio) if sigma_ratio else None,
175
+ 'sigma_deviation': float(sigma_deviation),
176
+ 'r_squared': float(r_squared),
177
+ 'composite_score': float(composite_score),
178
+ }
179
+
180
+ logger.debug(
181
+ f'[Fit Score] X0_err={x0_error:.4f}, sigma_ratio={sigma_ratio:.3f}, R_squared={r_squared:.4f}, Score={composite_score:.2f}'
182
+ )
183
+
184
+ return composite_score, metrics
185
+
186
+ except Exception as e:
187
+ logger.exception(f'[calculate_multi_parameter_fit_score] Exception: {str(e)}')
188
+ return 999.0, {'x0_error': 999.0, 'sigma_ratio': None, 'r_squared': None}
189
+
190
+ def match_isotope_pattern(
191
+ self,
192
+ experimental_mz: npt.NDArray[np.float64],
193
+ experimental_int: npt.NDArray[np.float64],
194
+ theoretical_pattern: dict,
195
+ tolerance: float = 0.5,
196
+ ) -> float:
197
+ """
198
+ Match experimental peaks to theoretical isotope pattern using Gaussian fitting
199
+ Compares X0 (centroid) positions between theory and experiment
200
+ Returns the X0 centroid difference in m/z units (error metric)
201
+ """
202
+ if 'error' in theoretical_pattern:
203
+ return 999.0 # Large error if pattern generation failed
204
+
205
+ # Use smooth Gaussian pattern for theo_x0 calculation (same method as exp_x0)
206
+ theo_mz = np.array(theoretical_pattern.get('gaussian_mz', theoretical_pattern.get('mz', [])))
207
+ theo_int = np.array(theoretical_pattern.get('gaussian_intensity', theoretical_pattern.get('intensity', [])))
208
+
209
+ if len(theo_mz) == 0:
210
+ return 999.0
211
+
212
+ # Normalize both patterns
213
+ theo_int_norm = theo_int / np.max(theo_int) * 100
214
+ exp_int_norm = experimental_int / np.max(experimental_int) * 100
215
+
216
+ # Calculate Gaussian centroids (X0) for both patterns
217
+ theo_fit_result = self.gaussian_fit_centroid(theo_mz, theo_int_norm)
218
+ exp_fit_result = self.gaussian_fit_centroid(experimental_mz, exp_int_norm)
219
+
220
+ theo_x0 = theo_fit_result[0] if theo_fit_result else None
221
+ exp_x0 = exp_fit_result[0] if exp_fit_result else None
222
+
223
+ if theo_x0 is None or exp_x0 is None:
224
+ return 999.0
225
+
226
+ # Find experimental peak closest to each theoretical peak
227
+ matched_intensities = []
228
+ matched_masses = []
229
+
230
+ for t_mz, t_int in zip(theo_mz, theo_int_norm):
231
+ # Find experimental peaks within tolerance
232
+ close_peaks = np.where(np.abs(experimental_mz - t_mz) < tolerance)[0]
233
+
234
+ if len(close_peaks) > 0:
235
+ # Find the closest peak
236
+ closest_idx = close_peaks[np.argmin(np.abs(experimental_mz[close_peaks] - t_mz))]
237
+ matched_intensities.append((t_int, exp_int_norm[closest_idx]))
238
+ matched_masses.append((t_mz, experimental_mz[closest_idx]))
239
+
240
+ if len(matched_intensities) == 0:
241
+ return 999.0
242
+
243
+ # Return X0 centroid difference in m/z units
244
+ # This is the ERROR metric - smaller is better
245
+ # Different Qcl values shift the centroid position
246
+ # The Qcl with smallest X0 error is the correct one
247
+
248
+ x0_error = abs(exp_x0 - theo_x0)
249
+
250
+ return x0_error
core/spectrum.py ADDED
@@ -0,0 +1,293 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import logging
4
+ import os
5
+ import sys
6
+ from typing import Optional
7
+
8
+ import numpy as np
9
+ import numpy.typing as npt
10
+
11
+ current_dir = os.path.dirname(os.path.abspath(__file__))
12
+ sys.path.insert(0, os.path.join(current_dir, '..', 'lib'))
13
+
14
+ from pythoms.tome import autoresolution
15
+
16
+ logger = logging.getLogger(__name__)
17
+
18
+
19
+ class SpectrumMixin:
20
+ """Mixin for spectrum parsing and peak detection methods."""
21
+
22
+ def parse_txt_spectrum(self, txt_content: str) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
23
+ """
24
+ Parse mass spectrum from txt file
25
+ Expected format: two columns (m/z, intensity) separated by whitespace or comma
26
+ """
27
+ lines = txt_content.strip().replace('\r\n', '\n').replace('\r', '\n').split('\n')
28
+ mz_values = []
29
+ intensity_values = []
30
+
31
+ for line in lines:
32
+ line = line.strip()
33
+ if not line or line.startswith('#'):
34
+ continue
35
+
36
+ # Try different delimiters
37
+ parts = None
38
+ if '\t' in line:
39
+ parts = line.split('\t')
40
+ elif ',' in line:
41
+ parts = line.split(',')
42
+ else:
43
+ parts = line.split()
44
+
45
+ if len(parts) >= 2:
46
+ try:
47
+ mz = float(parts[0])
48
+ intensity = float(parts[1])
49
+ mz_values.append(mz)
50
+ intensity_values.append(intensity)
51
+ except ValueError:
52
+ continue
53
+
54
+ return np.array(mz_values), np.array(intensity_values)
55
+
56
+ def estimate_resolution(self, mz_values: npt.NDArray[np.float64], intensity_values: npt.NDArray[np.float64]) -> int:
57
+ """
58
+ Estimate the resolution of the spectrum using PythoMS methodology
59
+ """
60
+ try:
61
+ res = autoresolution(list(mz_values), list(intensity_values), n=10, v=False)
62
+ if res is None or not np.isfinite(res) or res <= 0:
63
+ return 20000 # Default resolution
64
+ return int(res)
65
+ except Exception:
66
+ return 20000 # Default resolution if estimation fails
67
+
68
+ def calculate_fwhm(self, mz: float, resolution: int) -> float:
69
+ """
70
+ Calculate Full Width at Half Maximum for a given m/z and resolution
71
+ FWHM = m/z / resolution
72
+ """
73
+ return mz / resolution
74
+
75
+ def find_local_maximum(
76
+ self,
77
+ mz_values: npt.NDArray[np.float64],
78
+ intensity_values: npt.NDArray[np.float64],
79
+ center_mz: float,
80
+ lookwithin: Optional[float] = None,
81
+ ) -> tuple[Optional[float], Optional[float]]:
82
+ """
83
+ Find the local maximum within a window around center_mz
84
+ Based on PythoMS localmax function
85
+ """
86
+ if lookwithin is None:
87
+ lookwithin = 1.0
88
+
89
+ # Find indices within the window
90
+ left_idx = np.searchsorted(mz_values, center_mz - lookwithin, side='left')
91
+ right_idx = np.searchsorted(mz_values, center_mz + lookwithin, side='right')
92
+
93
+ if left_idx >= right_idx:
94
+ return None, None
95
+
96
+ # Find maximum in the window
97
+ window_intensities = intensity_values[left_idx:right_idx]
98
+ if len(window_intensities) == 0:
99
+ return None, None
100
+
101
+ max_intensity = np.max(window_intensities)
102
+ max_idx_in_window = np.argmax(window_intensities)
103
+ max_idx = left_idx + max_idx_in_window
104
+
105
+ return mz_values[max_idx], max_intensity
106
+
107
+ def find_peak_regions(
108
+ self,
109
+ mz_values: npt.NDArray[np.float64],
110
+ intensity_values: npt.NDArray[np.float64],
111
+ threshold: float = 0.05,
112
+ merge_gap: float = 1.5,
113
+ ) -> list[tuple[int, int]]:
114
+ """
115
+ Find isotope envelope regions - each ENVELOPE is one region
116
+ Merges nearby regions that are likely part of the same isotope envelope
117
+ merge_gap: merge regions separated by less than this m/z (default 1.5)
118
+ """
119
+ norm_intensity = intensity_values / np.max(intensity_values)
120
+
121
+ # Find regions above threshold
122
+ above_threshold = norm_intensity > threshold
123
+
124
+ # Find all continuous regions above threshold
125
+ regions = []
126
+ in_region = False
127
+ start_idx = 0
128
+
129
+ for i in range(len(above_threshold)):
130
+ if above_threshold[i] and not in_region:
131
+ # Start of new region
132
+ start_idx = i
133
+ in_region = True
134
+ elif not above_threshold[i] and in_region:
135
+ # End of region
136
+ end_idx = i - 1
137
+ regions.append((start_idx, end_idx))
138
+ in_region = False
139
+
140
+ # Handle case where last region extends to end
141
+ if in_region:
142
+ regions.append((start_idx, len(above_threshold) - 1))
143
+
144
+ # Merge regions that are close together (likely same isotope envelope)
145
+ if len(regions) <= 1:
146
+ return regions
147
+
148
+ merged_regions = []
149
+ current_start, current_end = regions[0]
150
+
151
+ for i in range(1, len(regions)):
152
+ next_start, next_end = regions[i]
153
+
154
+ # Check gap between current region end and next region start
155
+ gap = mz_values[next_start] - mz_values[current_end]
156
+
157
+ if gap < merge_gap:
158
+ # Merge: extend current region to include next region
159
+ current_end = next_end
160
+ else:
161
+ # Don't merge: save current region and start new one
162
+ merged_regions.append((current_start, current_end))
163
+ current_start, current_end = next_start, next_end
164
+
165
+ # Add the last region
166
+ merged_regions.append((current_start, current_end))
167
+
168
+ return merged_regions
169
+
170
+ def detect_peak_boundaries(
171
+ self, mz_array: npt.NDArray[np.float64], int_array: npt.NDArray[np.float64], peak_mz: float
172
+ ) -> tuple[float, float, float]:
173
+ """
174
+ Detect the boundaries of a single isotope envelope from experimental data.
175
+
176
+ NEW APPROACH:
177
+ 1. Find APEX (highest point) in small window around clicked position
178
+ 2. From apex, scan left/right to find valleys (local minima)
179
+ 3. This ensures we identify the correct peak without jumping to adjacent ones
180
+
181
+ Returns: (left_boundary_mz, right_boundary_mz, apex_mz)
182
+ """
183
+ # Find the index closest to the clicked peak
184
+ peak_idx = np.argmin(np.abs(mz_array - peak_mz))
185
+
186
+ # STEP 1: Find the APEX (highest point) in a SMALL window around clicked position
187
+ # Use small window (±10 points) to avoid jumping to adjacent peaks
188
+ search_window = 10 # Small window to stay on same peak
189
+ search_start = max(0, peak_idx - search_window)
190
+ search_end = min(len(int_array), peak_idx + search_window)
191
+
192
+ # Find apex within this small region
193
+ local_region_intensities = int_array[search_start:search_end]
194
+ apex_idx_in_region = np.argmax(local_region_intensities)
195
+ apex_idx = search_start + apex_idx_in_region
196
+
197
+ apex_mz = mz_array[apex_idx]
198
+ apex_intensity = int_array[apex_idx]
199
+
200
+ logger.debug(f'Detecting isotope envelope boundaries around clicked m/z={peak_mz:.4f}')
201
+ logger.debug(f'Clicked at index={peak_idx}, mz={mz_array[peak_idx]:.4f}')
202
+ logger.debug(f'Found APEX at index={apex_idx}, mz={apex_mz:.4f}, intensity={apex_intensity:.0f}')
203
+
204
+ # STEP 2: From APEX, scan LEFT to find valley (local minimum)
205
+ left_idx = apex_idx
206
+ min_intensity_left = apex_intensity
207
+
208
+ for i in range(apex_idx - 1, max(0, apex_idx - 200), -1):
209
+ current_intensity = int_array[i]
210
+
211
+ # Track the minimum intensity as we scan left
212
+ if current_intensity < min_intensity_left:
213
+ min_intensity_left = current_intensity
214
+ left_idx = i
215
+
216
+ # Stop if intensity starts rising significantly (found the valley)
217
+ # Look for 2 consecutive points rising by >10%
218
+ if i >= 1:
219
+ if int_array[i - 1] > current_intensity * 1.1 and int_array[i] > int_array[i + 1] * 1.1:
220
+ # Found a valley - intensity is rising on the left
221
+ logger.debug(
222
+ f'Left valley at index={left_idx}, mz={mz_array[left_idx]:.4f}, intensity={int_array[left_idx]:.0f}'
223
+ )
224
+ break
225
+
226
+ # If we hit the edge without finding a valley, use the minimum we found
227
+ if left_idx == apex_idx:
228
+ logger.debug(f'Left boundary at edge: index={left_idx}, mz={mz_array[left_idx]:.4f}')
229
+
230
+ # STEP 3: From APEX, scan RIGHT to find valley (local minimum)
231
+ right_idx = apex_idx
232
+ min_intensity_right = apex_intensity
233
+
234
+ for i in range(apex_idx + 1, min(len(int_array), apex_idx + 200)):
235
+ current_intensity = int_array[i]
236
+
237
+ # Track the minimum intensity as we scan right
238
+ if current_intensity < min_intensity_right:
239
+ min_intensity_right = current_intensity
240
+ right_idx = i
241
+
242
+ # Stop if intensity starts rising significantly (found the valley)
243
+ # Look for 2 consecutive points rising by >10%
244
+ if i < len(int_array) - 1:
245
+ if int_array[i + 1] > current_intensity * 1.1 and int_array[i] > int_array[i - 1] * 1.1:
246
+ # Found a valley - intensity is rising on the right
247
+ logger.debug(
248
+ f'Right valley at index={right_idx}, mz={mz_array[right_idx]:.4f}, intensity={int_array[right_idx]:.0f}'
249
+ )
250
+ break
251
+
252
+ # If we hit the edge without finding a valley, use the minimum we found
253
+ if right_idx == apex_idx:
254
+ logger.debug(f'Right boundary at edge: index={right_idx}, mz={mz_array[right_idx]:.4f}')
255
+
256
+ left_boundary_mz = mz_array[left_idx]
257
+ right_boundary_mz = mz_array[right_idx]
258
+ width = right_boundary_mz - left_boundary_mz
259
+ num_points = right_idx - left_idx + 1
260
+
261
+ logger.debug(
262
+ f'Final envelope: [{left_boundary_mz:.4f}, {right_boundary_mz:.4f}] m/z (width={width:.4f}, {num_points} points)'
263
+ )
264
+ logger.debug(f'Apex at {apex_mz:.4f} (Gaussian will use MIDPOINT of boundaries as initial guess)')
265
+
266
+ return left_boundary_mz, right_boundary_mz, apex_mz
267
+
268
+ def weighted_centroid(
269
+ self,
270
+ mz_values: npt.NDArray[np.float64],
271
+ intensity_values: npt.NDArray[np.float64],
272
+ start_idx: int,
273
+ end_idx: int,
274
+ ) -> tuple[Optional[float], Optional[float]]:
275
+ """
276
+ Calculate peak centroid (position of maximum intensity) matching PythoMS isotope overlay method
277
+
278
+ This uses the m/z value at maximum intensity for peak position, which is consistent
279
+ with how PythoMS's plot_mass_spectrum and localmax functions work.
280
+ """
281
+ region_mz = mz_values[start_idx : end_idx + 1]
282
+ region_int = intensity_values[start_idx : end_idx + 1]
283
+
284
+ if len(region_mz) == 0 or np.sum(region_int) == 0:
285
+ return None, None
286
+
287
+ # Find the m/z at maximum intensity (peak apex)
288
+ # This matches PythoMS isotope overlay behavior
289
+ max_idx = np.argmax(region_int)
290
+ centroid_mz = region_mz[max_idx]
291
+ max_intensity = region_int[max_idx]
292
+
293
+ return centroid_mz, max_intensity
dna_silver_webapp.py CHANGED
The diff for this file is too large to render. See raw diff
 
sample_data/DNAdup_20250421_dC20_1p5Ag.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/GG322-BCN.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/GG322.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/SNA-C10_AgN.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/TNA-C10_AgN.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/XNAdup_SNA_C10.txt ADDED
The diff for this file is too large to render. See raw diff
 
sample_data/XNAdup_TNA_C10.txt ADDED
The diff for this file is too large to render. See raw diff
 
templates/index.html CHANGED
@@ -2696,7 +2696,7 @@
2696
  <div style="display: flex; align-items: center; gap: 10px;">
2697
  <span style="font-size: 1.5em;">✅</span>
2698
  <div style="flex: 1;">
2699
- <strong style="color: #2e7d32; font-size: 1.05em;">Check boxes to overlay theoritical spectra</strong>
2700
  <div style="font-size: 0.9em; color: #555; margin-top: 4px;">
2701
  Toggle checkboxes to compare theoretical isotope patterns with experimental data on both main and zoomed plots
2702
  </div>
 
2696
  <div style="display: flex; align-items: center; gap: 10px;">
2697
  <span style="font-size: 1.5em;">✅</span>
2698
  <div style="flex: 1;">
2699
+ <strong style="color: #2e7d32; font-size: 1.05em;">Check boxes to overlay theoretical spectra</strong>
2700
  <div style="font-size: 0.9em; color: #555; margin-top: 4px;">
2701
  Toggle checkboxes to compare theoretical isotope patterns with experimental data on both main and zoomed plots
2702
  </div>