corems.mass_spectrum.calc.Calibration

Created on Wed May 13 02:16:09 2020

@author: Will Kew

  1# -*- coding: utf-8 -*-
  2"""
  3Created on Wed May 13 02:16:09 2020
  4
  5@author: Will Kew
  6"""
  7
  8# import modules
  9import csv
 10import warnings
 11from io import BytesIO, StringIO
 12from pathlib import Path
 13
 14import numpy as np
 15import pandas as pd
 16from s3path import S3Path
 17
 18# import scipy modules for calibration
 19from scipy.optimize import minimize
 20
 21
 22class MzDomainCalibration:
 23    """MzDomainCalibration class for recalibrating mass spectra
 24
 25    Parameters
 26    ----------
 27    mass_spectrum : CoreMS MassSpectrum Object
 28        The mass spectrum to be calibrated.
 29    ref_masslist : str
 30        The path to a reference mass list.
 31    mzsegment : tuple of floats, optional
 32        The mz range to recalibrate, or None. Used for calibration of specific parts of the mz domain at a time.
 33        Future work - allow multiple mzsegments to be passed.
 34
 35    Attributes
 36    ----------
 37    mass_spectrum : CoreMS MassSpectrum Object
 38        The mass spectrum to be calibrated.
 39    mzsegment : tuple of floats or None
 40        The mz range to recalibrate, or None.
 41    ref_mass_list_path : str or Path
 42        The path to the reference mass list.
 43
 44    Methods
 45    -------
 46    * run().
 47        Main function to run this class.
 48    * load_ref_mass_list().
 49        Load reference mass list (Bruker format).
 50    * gen_ref_mass_list_from_assigned(min_conf=0.7).
 51        Generate reference mass list from assigned masses.
 52    * find_calibration_points(df_ref, calib_ppm_error_threshold=(-1, 1), calib_snr_threshold=5).
 53        Find calibration points in the mass spectrum based on the reference mass list.
 54    * robust_calib(param, cal_peaks_mz, cal_refs_mz, order=1).
 55        Recalibration function.
 56    * recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=1, diagnostic=False).
 57        Main recalibration function which uses a robust linear regression.
 58
 59
 60    """
 61
 62    def __init__(self, mass_spectrum, ref_masslist, mzsegment=None):
 63        self.mass_spectrum = mass_spectrum
 64        self.mzsegment = mzsegment
 65
 66        # define reference mass list - bruker .ref format
 67        self.ref_mass_list_path = ref_masslist
 68        if self.mass_spectrum.percentile_assigned(mute_output=True)[0] != 0:
 69            warnings.warn(
 70                "Warning: calibrating spectra which have already been assigned may yield erroneous results"
 71            )
 72        self.mass_spectrum.mz_cal = self.mass_spectrum.mz_exp
 73        self.mass_spectrum.mz_cal_profile = self.mass_spectrum._mz_exp
 74
 75        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
 76            print(
 77                "MS Obj loaded - " + str(len(mass_spectrum.mspeaks)) + " peaks found."
 78            )
 79     
 80    def load_ref_mass_list(self):
 81        """
 82        Load reference mass list (Bruker format).
 83
 84        Loads in a reference mass list from a .ref file. Some versions of
 85        Bruker's software produce .ref files with a different format where
 86        the header lines (starting with '#' or '##') and delimiters may vary.
 87        The file may be located locally or on S3 and will be handled accordingly.
 88        
 89        Returns
 90        -------
 91        df_ref : Pandas DataFrame
 92            Reference mass list object.
 93        """
 94        # Get a Path-like object from the input path string or S3Path
 95        refmasslist = (Path(self.ref_mass_list_path)
 96                    if isinstance(self.ref_mass_list_path, str)
 97                    else self.ref_mass_list_path)
 98        
 99        # Make sure the file exists
100        if not refmasslist.exists():
101            raise FileExistsError("File does not exist: %s" % refmasslist)
102        
103        # Read all lines from the file (handling S3 vs local differently)
104        if isinstance(refmasslist, S3Path):
105            # For S3, read the file in binary, then decode to string and split into lines.
106            content = refmasslist.open("rb").read()
107            all_lines = content.decode("utf-8").splitlines(keepends=True)
108        else:
109            # For a local file, open in text mode and read lines.
110            with refmasslist.open("r") as f:
111                all_lines = f.readlines()
112        
113        # Identify the index of the first line of the actual data.
114        # We assume header lines start with '#' (or '##') and ignore blank lines.
115        data_start_idx = 0
116        for idx, line in enumerate(all_lines):
117            if line.strip() and not line.lstrip().startswith("#"):
118                data_start_idx = idx
119                break
120        
121        # If there are not enough lines to guess the dialect, throw an error
122        if data_start_idx >= len(all_lines):
123            raise ValueError("The file does not appear to contain any data lines.")
124        
125        # Use a couple of the data lines to let csv.Sniffer detect the delimiter
126        sample_lines = "".join(all_lines[data_start_idx:data_start_idx+2])
127        try:
128            dialect = csv.Sniffer().sniff(sample_lines)
129            delimiter = dialect.delimiter
130        except csv.Error:
131            # If csv.Sniffer fails, default to a common delimiter (e.g., comma)
132            delimiter = ","
133        
134        # Join the lines from the beginning of data (this might include a blank line) 
135        joined_data = "".join(all_lines[data_start_idx:])
136        
137        # Depending on whether the file is S3 or local, wrap the data as needed for pandas
138        if isinstance(refmasslist, S3Path):
139            data_buffer = BytesIO(joined_data.encode("utf-8"))
140        else:
141            data_buffer = StringIO(joined_data)
142        
143        # Read data into a DataFrame.
144        # Adjust columns and names as needed – here we assume at least 2 columns:
145        df_ref = pd.read_csv(data_buffer,
146                            sep=delimiter,
147                            header=None,
148                            usecols=[0, 1],   # Modify if more columns are required.
149                            names=["Formula", "m/z"])
150        
151        df_ref.sort_values(by="m/z", ascending=True, inplace=True)
152        
153        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
154            print("Reference mass list loaded - {} calibration masses loaded.".format(len(df_ref)))
155        
156        return df_ref
157
158    def gen_ref_mass_list_from_assigned(self, min_conf: float = 0.7):
159        """Generate reference mass list from assigned masses
160
161        This function will generate a ref mass dataframe object from an assigned corems mass spec obj
162        using assigned masses above a certain minimum confidence threshold.
163
164        This function needs to be retested and check it is covered in the unit tests.
165
166        Parameters
167        ----------
168        min_conf : float, optional
169            minimum confidence score. The default is 0.7.
170
171        Returns
172        -------
173        df_ref : Pandas DataFrame
174            reference mass list - based on calculated masses.
175
176        """
177        # TODO this function needs to be retested and check it is covered in the unit tests
178        df = self.mass_spectrum.to_dataframe()
179        df = df[df["Confidence Score"] > min_conf]
180        df_ref = pd.DataFrame(columns=["m/z"])
181        df_ref["m/z"] = df["Calculated m/z"]
182        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
183            print(
184                "Reference mass list generated - "
185                + str(len(df_ref))
186                + " calibration masses."
187            )
188        return df_ref
189
190    def find_calibration_points(
191        self,
192        df_ref,
193        calib_ppm_error_threshold: tuple[float, float] = (-1, 1),
194        calib_snr_threshold: float = 5,
195        calibration_ref_match_method: str = "merged",
196        calibration_ref_match_tolerance: float = 0.003,
197        calibration_ref_match_std_raw_error_limit: float = 1.5,
198    ):
199        """Function to find calibration points in the mass spectrum
200
201        Based on the reference mass list.
202
203        Parameters
204        ----------
205        df_ref : Pandas DataFrame
206            reference mass list for recalibration.
207        calib_ppm_error_threshold : tuple of floats, optional
208            ppm error for finding calibration masses in the spectrum. The default is -1,1.
209            Note: This is based on the calculation of ppm = ((mz_measure - mz_theoretical)/mz_theoretical)*1e6.
210                Some software does this the other way around and value signs must be inverted for that to work.
211        calib_snr_threshold : float, optional
212            snr threshold for finding calibration masses in the spectrum. The default is 5.
213            If SNR data is unavailable, peaks are filtered by intensity percentile using the formula:
214            percentile = max(5, 100 - calib_snr_threshold)
215        calibration_ref_match_method : str, optional
216            method for matching calibration references. The default is "merged".
217        calibration_ref_match_tolerance : float, optional
218            tolerance for matching calibration references. The default is 0.003.
219        calibration_ref_match_std_raw_error_limit : float, optional
220            standard deviation raw error limit for calibration references. The default is 1.5.
221
222        Returns
223        -------
224        cal_peaks_mz : list of floats
225            masses of measured ions to use in calibration routine
226        cal_refs_mz : list of floats
227            reference mz values of found calibration points.
228
229        """
230
231        # Check if SNR data is available by testing the first peak
232        use_snr = False
233        if len(self.mass_spectrum.mspeaks) > 0:
234            first_peak = self.mass_spectrum.mspeaks[0]
235            if (hasattr(first_peak, 'signal_to_noise') and 
236                first_peak.signal_to_noise is not None and 
237                not np.isnan(first_peak.signal_to_noise) and
238                first_peak.signal_to_noise > 0):
239                use_snr = True
240
241        # This approach is much more efficient and expedient than the original implementation.
242        peaks_mz = []
243        peaks_intensity = []
244        
245        if use_snr:
246            # Use SNR filtering
247            for x in self.mass_spectrum.mspeaks:
248                if x.signal_to_noise > calib_snr_threshold:
249                    if self.mzsegment:
250                        if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
251                            peaks_mz.append(x.mz_exp)
252                    else:
253                        peaks_mz.append(x.mz_exp)
254        else:
255            # Fallback to intensity percentile filtering
256            intensity_percentile = max(5, 100 - calib_snr_threshold)
257            warnings.warn(
258                f"SNR data unavailable for calibration. Using intensity-based filtering instead. "
259                f"SNR threshold of {calib_snr_threshold} corresponds to intensity percentile >= {intensity_percentile}%."
260            )
261            
262            # Collect all peaks and their intensities
263            all_peaks_data = []
264            for x in self.mass_spectrum.mspeaks:
265                if self.mzsegment:
266                    if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
267                        all_peaks_data.append((x.mz_exp, x.abundance))
268                else:
269                    all_peaks_data.append((x.mz_exp, x.abundance))
270            
271            if all_peaks_data:
272                peaks_mz_list, intensities = zip(*all_peaks_data)
273                intensity_threshold = np.percentile(intensities, intensity_percentile)
274                
275                for mz, intensity in all_peaks_data:
276                    if intensity >= intensity_threshold:
277                        peaks_mz.append(mz)
278        
279        peaks_mz = np.asarray(peaks_mz)
280
281        if calibration_ref_match_method == "legacy":
282            # This legacy approach iterates through each reference match and finds the entries within 1 mz and within the user defined PPM error threshold
283            # Then it removes ambiguities - which means the calibration threshold hasto be very tight.
284            cal_peaks_mz = []
285            cal_refs_mz = []
286            for mzref in df_ref["m/z"]:
287                tmp_peaks_mz = peaks_mz[abs(peaks_mz - mzref) < 1]
288                for mzmeas in tmp_peaks_mz:
289                    delta_mass = ((mzmeas - mzref) / mzref) * 1e6
290                    if delta_mass < max(calib_ppm_error_threshold):
291                        if delta_mass > min(calib_ppm_error_threshold):
292                            cal_peaks_mz.append(mzmeas)
293                            cal_refs_mz.append(mzref)
294
295            # To remove entries with duplicated indices (reference masses matching multiple peaks)
296            tmpdf = pd.Series(index=cal_refs_mz, data=cal_peaks_mz, dtype=float)
297            tmpdf = tmpdf[~tmpdf.index.duplicated(keep=False)]
298
299            cal_peaks_mz = list(tmpdf.values)
300            cal_refs_mz = list(tmpdf.index)
301        elif calibration_ref_match_method == "merged":
302            # This is a new approach (August 2024) which uses Pandas 'merged_asof' to find the peaks closest in m/z between
303            # reference and measured masses. This is a quicker way to match, and seems to get more matches.
304            # It may not work as well when the data are far from correc initial mass
305            # e.g. if the correct peak is further from the reference than an incorrect peak.
306            meas_df = pd.DataFrame(columns=["meas_m/z"], data=peaks_mz)
307            tolerance = calibration_ref_match_tolerance
308            merged_df = pd.merge_asof(
309                df_ref,
310                meas_df,
311                left_on="m/z",
312                right_on="meas_m/z",
313                tolerance=tolerance,
314                direction="nearest",
315            )
316            merged_df.dropna(how="any", inplace=True)
317            merged_df["Error_ppm"] = (
318                (merged_df["meas_m/z"] - merged_df["m/z"]) / merged_df["m/z"]
319            ) * 1e6
320            median_raw_error = merged_df["Error_ppm"].median()
321            std_raw_error = merged_df["Error_ppm"].std()
322            if std_raw_error > calibration_ref_match_std_raw_error_limit:
323                std_raw_error = calibration_ref_match_std_raw_error_limit
324            self.mass_spectrum.calibration_raw_error_median = median_raw_error
325            self.mass_spectrum.calibration_raw_error_stdev = std_raw_error
326            merged_df = merged_df[
327                (merged_df["Error_ppm"] > (median_raw_error - 1.5 * std_raw_error))
328                & (merged_df["Error_ppm"] < (median_raw_error + 1.5 * std_raw_error))
329            ]
330            # merged_df= merged_df[(merged_df['Error_ppm']>min(calib_ppm_error_threshold))&(merged_df['Error_ppm']<max(calib_ppm_error_threshold))]
331            cal_peaks_mz = list(merged_df["meas_m/z"])
332            cal_refs_mz = list(merged_df["m/z"])
333        else:
334            raise ValueError(f"{calibration_ref_match_method} not allowed.")
335
336        # it is crucial the mass lists are in same order
337        # corems likes to do masses from high to low.
338        cal_refs_mz.sort(reverse=False)
339        cal_peaks_mz.sort(reverse=False)
340        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
341            print(
342                str(len(cal_peaks_mz))
343                + " calibration points matched within thresholds."
344            )
345        return cal_peaks_mz, cal_refs_mz
346
347    def robust_calib(
348        self,
349        param: list[float],
350        cal_peaks_mz: list[float],
351        cal_refs_mz: list[float],
352        order: int = 1,
353    ):
354        """Recalibration function
355
356        Computes the rms of m/z errors to minimize when calibrating.
357        This is adapted from from spike.
358
359        Parameters
360        ----------
361        param : list of floats
362            generated by minimize function from scipy optimize.
363        cal_peaks_mz : list of floats
364            masses of measured peaks to use in mass calibration.
365        cal_peaks_mz : list of floats
366            reference mz values of found calibration points.
367        order : int, optional
368            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
369
370        Returns
371        -------
372        rmserror : float
373            root mean square mass error for calibration points.
374
375        """
376        Aterm = param[0]
377        Bterm = param[1]
378        try:
379            Cterm = param[2]
380        except IndexError:
381            pass
382
383        # get the mspeaks from the mass spectrum object which were calibration points
384        # mspeaks = [self.mass_spectrum.mspeaks[x] for x in imzmeas]
385        # get their calibrated mass values
386        # mspeakmzs = [x.mz_cal for x in mspeaks]
387        cal_peaks_mz = np.asarray(cal_peaks_mz)
388
389        # linearz
390        if order == 1:
391            ref_recal_points = (Aterm * cal_peaks_mz) + Bterm
392        # quadratic
393        elif order == 2:
394            ref_recal_points = (Aterm * (cal_peaks_mz)) + (
395                Bterm * np.power((cal_peaks_mz), 2) + Cterm
396            )
397
398        # sort both the calibration points (measured, recalibrated)
399        ref_recal_points.sort()
400        # and sort the calibration points (theoretical, predefined)
401        cal_refs_mz.sort()
402
403        # calculate the ppm error for each calibration point
404        error = ((ref_recal_points - cal_refs_mz) / cal_refs_mz) * 1e6
405        # calculate the root mean square error - this is our target to minimize
406        rmserror = np.sqrt(np.mean(error**2))
407        return rmserror
408
409    def recalibrate_mass_spectrum(
410        self,
411        cal_peaks_mz: list[float],
412        cal_refs_mz: list[float],
413        order: int = 1,
414        diagnostic: bool = False,
415    ):
416        """Main recalibration function which uses a robust linear regression
417
418        This function performs the recalibration of the mass spectrum object.
419        It iteratively applies
420
421        Parameters
422        ----------
423        cal_peaks_mz : list of float
424            masses of measured peaks to use in mass calibration.
425        cal_refs_mz : list of float
426            reference mz values of found calibration points.
427        order : int, optional
428            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
429
430        Returns
431        -------
432        mass_spectrum : CoreMS mass spectrum object
433            Calibrated mass spectrum object
434
435
436        Notes
437        -----
438        This function is adapted, in part, from the SPIKE project [1,2] and is based on the robust linear regression method.
439
440        References
441        ----------
442        1.  Chiron L., Coutouly M-A., Starck J-P., Rolando C., Delsuc M-A.
443            SPIKE a Processing Software dedicated to Fourier Spectroscopies
444            https://arxiv.org/abs/1608.06777 (2016)
445        2.  SPIKE - https://github.com/spike-project/spike
446
447        """
448        # initialise parameters for recalibration
449        # these are the 'Aterm, Bterm, Cterm'
450        # as spectra are already freq->mz calibrated, these terms are very small
451        # may be beneficial to formally separate them from the freq->mz terms
452        if order == 1:
453            Po = [1, 0]
454        elif order == 2:
455            Po = [1, 0, 0]
456
457        if len(cal_peaks_mz) >= 2:
458            if self.mzsegment:  # If only part of the spectrum is to be recalibrated
459                mz_exp_peaks = np.array(
460                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
461                )
462                # Split the array into two parts - one to recailbrate, one to keep unchanged.
463                mz_exp_peaks_tocal = mz_exp_peaks[
464                    (mz_exp_peaks >= min(self.mzsegment))
465                    & (mz_exp_peaks <= max(self.mzsegment))
466                ]
467                mz_exp_peaks_unchanged = mz_exp_peaks[
468                    ~(mz_exp_peaks >= min(self.mzsegment))
469                    | ~(mz_exp_peaks <= max(self.mzsegment))
470                ]
471                # TODO: - segmented calibration needs a way to better track the calibration args/values...
472                if not self.mass_spectrum.is_centroid:
473                    mz_exp_profile = np.array(self.mass_spectrum.mz_exp_profile)
474                    # Split the array into two parts - one to recailbrate, one to keep unchanged.
475                    mz_exp_profile_tocal = mz_exp_profile[
476                        (mz_exp_profile >= min(self.mzsegment))
477                        & (mz_exp_profile <= max(self.mzsegment))
478                    ]
479                    mz_exp_profile_unchanged = mz_exp_profile[
480                        ~(mz_exp_profile >= min(self.mzsegment))
481                        | ~(mz_exp_profile <= max(self.mzsegment))
482                    ]
483            else:  # if just recalibrating the whole spectrum
484                mz_exp_peaks_tocal = np.array(
485                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
486                )
487                if not self.mass_spectrum.is_centroid:
488                    mz_exp_profile_tocal = np.array(self.mass_spectrum.mz_exp_profile)
489
490            minimize_method = self.mass_spectrum.settings.calib_minimize_method
491            res = minimize(
492                self.robust_calib,
493                Po,
494                args=(cal_peaks_mz, cal_refs_mz, order),
495                method=minimize_method,
496            )
497            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
498                print(
499                    "minimize function completed with RMS error of: {:0.3f} ppm".format(
500                        res["fun"]
501                    )
502                )
503                print(
504                    "minimize function performed {:1d} fn evals and {:1d} iterations".format(
505                        res["nfev"], res["nit"]
506                    )
507                )
508            Pn = res.x
509
510            # mz_exp_ms = np.array([mspeak.mz_exp for mspeak in self.mass_spectrum])
511
512            if order == 1:
513                mz_domain = (Pn[0] * mz_exp_peaks_tocal) + Pn[1]
514                if not self.mass_spectrum.is_centroid:
515                    mz_profile_calc = (Pn[0] * mz_exp_profile_tocal) + Pn[1]
516
517            elif order == 2:
518                mz_domain = (Pn[0] * (mz_exp_peaks_tocal)) + (
519                    Pn[1] * np.power((mz_exp_peaks_tocal), 2) + Pn[2]
520                )
521
522                if not self.mass_spectrum.is_centroid:
523                    mz_profile_calc = (Pn[0] * (mz_exp_profile_tocal)) + (
524                        Pn[1] * np.power((mz_exp_profile_tocal), 2) + Pn[2]
525                    )
526
527            if self.mzsegment:
528                # Recombine the mass domains
529                mz_domain = np.concatenate([mz_domain, mz_exp_peaks_unchanged])
530                mz_domain.sort()
531                if not self.mass_spectrum.is_centroid:
532                    mz_profile_calc = np.concatenate(
533                        [mz_profile_calc, mz_exp_profile_unchanged]
534                    )
535                    mz_profile_calc.sort()
536                # Sort them
537                if (
538                    mz_exp_peaks[0] > mz_exp_peaks[1]
539                ):  # If originally descending mass order
540                    mz_domain = mz_domain[::-1]
541                    if not self.mass_spectrum.is_centroid:
542                        mz_profile_calc = mz_profile_calc[::-1]
543
544            self.mass_spectrum.mz_cal = mz_domain
545            if not self.mass_spectrum.is_centroid:
546                self.mass_spectrum.mz_cal_profile = mz_profile_calc
547
548            self.mass_spectrum.calibration_order = order
549            self.mass_spectrum.calibration_RMS = float(res["fun"])
550            self.mass_spectrum.calibration_points = int(len(cal_refs_mz))
551            self.mass_spectrum.calibration_ref_mzs = cal_refs_mz
552            self.mass_spectrum.calibration_meas_mzs = cal_peaks_mz
553
554            self.mass_spectrum.calibration_segment = self.mzsegment
555
556            if diagnostic:
557                return self.mass_spectrum, res
558            return self.mass_spectrum
559        else:
560            warnings.warn("Too few calibration points - aborting.")
561            return self.mass_spectrum
562
563    def run(self):
564        """Run the calibration routine
565
566        This function runs the calibration routine.
567
568        """
569        calib_snr_threshold = self.mass_spectrum.settings.calib_sn_threshold
570        max_calib_ppm_error = self.mass_spectrum.settings.max_calib_ppm_error
571        min_calib_ppm_error = self.mass_spectrum.settings.min_calib_ppm_error
572        calib_pol_order = self.mass_spectrum.settings.calib_pol_order
573        calibration_ref_match_method = (
574            self.mass_spectrum.settings.calibration_ref_match_method
575        )
576        calibration_ref_match_tolerance = (
577            self.mass_spectrum.settings.calibration_ref_match_tolerance
578        )
579        calibration_ref_match_std_raw_error_limit = (
580            self.mass_spectrum.settings.calibration_ref_match_std_raw_error_limit
581        )
582
583        # load reference mass list
584        df_ref = self.load_ref_mass_list()
585
586        # find calibration points
587        cal_peaks_mz, cal_refs_mz = self.find_calibration_points(
588            df_ref,
589            calib_ppm_error_threshold=(min_calib_ppm_error, max_calib_ppm_error),
590            calib_snr_threshold=calib_snr_threshold,
591            calibration_ref_match_method=calibration_ref_match_method,
592            calibration_ref_match_tolerance=calibration_ref_match_tolerance,
593            calibration_ref_match_std_raw_error_limit=calibration_ref_match_std_raw_error_limit,
594        )
595        if len(cal_peaks_mz) == 2:
596            self.mass_spectrum.settings.calib_pol_order = 1
597            calib_pol_order = 1
598            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
599                print("Only 2 calibration points found, forcing a linear recalibration")
600        elif len(cal_peaks_mz) < 2:
601            warnings.warn("Too few calibration points found, function will fail")
602        self.recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=calib_pol_order)
class MzDomainCalibration:
 23class MzDomainCalibration:
 24    """MzDomainCalibration class for recalibrating mass spectra
 25
 26    Parameters
 27    ----------
 28    mass_spectrum : CoreMS MassSpectrum Object
 29        The mass spectrum to be calibrated.
 30    ref_masslist : str
 31        The path to a reference mass list.
 32    mzsegment : tuple of floats, optional
 33        The mz range to recalibrate, or None. Used for calibration of specific parts of the mz domain at a time.
 34        Future work - allow multiple mzsegments to be passed.
 35
 36    Attributes
 37    ----------
 38    mass_spectrum : CoreMS MassSpectrum Object
 39        The mass spectrum to be calibrated.
 40    mzsegment : tuple of floats or None
 41        The mz range to recalibrate, or None.
 42    ref_mass_list_path : str or Path
 43        The path to the reference mass list.
 44
 45    Methods
 46    -------
 47    * run().
 48        Main function to run this class.
 49    * load_ref_mass_list().
 50        Load reference mass list (Bruker format).
 51    * gen_ref_mass_list_from_assigned(min_conf=0.7).
 52        Generate reference mass list from assigned masses.
 53    * find_calibration_points(df_ref, calib_ppm_error_threshold=(-1, 1), calib_snr_threshold=5).
 54        Find calibration points in the mass spectrum based on the reference mass list.
 55    * robust_calib(param, cal_peaks_mz, cal_refs_mz, order=1).
 56        Recalibration function.
 57    * recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=1, diagnostic=False).
 58        Main recalibration function which uses a robust linear regression.
 59
 60
 61    """
 62
 63    def __init__(self, mass_spectrum, ref_masslist, mzsegment=None):
 64        self.mass_spectrum = mass_spectrum
 65        self.mzsegment = mzsegment
 66
 67        # define reference mass list - bruker .ref format
 68        self.ref_mass_list_path = ref_masslist
 69        if self.mass_spectrum.percentile_assigned(mute_output=True)[0] != 0:
 70            warnings.warn(
 71                "Warning: calibrating spectra which have already been assigned may yield erroneous results"
 72            )
 73        self.mass_spectrum.mz_cal = self.mass_spectrum.mz_exp
 74        self.mass_spectrum.mz_cal_profile = self.mass_spectrum._mz_exp
 75
 76        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
 77            print(
 78                "MS Obj loaded - " + str(len(mass_spectrum.mspeaks)) + " peaks found."
 79            )
 80     
 81    def load_ref_mass_list(self):
 82        """
 83        Load reference mass list (Bruker format).
 84
 85        Loads in a reference mass list from a .ref file. Some versions of
 86        Bruker's software produce .ref files with a different format where
 87        the header lines (starting with '#' or '##') and delimiters may vary.
 88        The file may be located locally or on S3 and will be handled accordingly.
 89        
 90        Returns
 91        -------
 92        df_ref : Pandas DataFrame
 93            Reference mass list object.
 94        """
 95        # Get a Path-like object from the input path string or S3Path
 96        refmasslist = (Path(self.ref_mass_list_path)
 97                    if isinstance(self.ref_mass_list_path, str)
 98                    else self.ref_mass_list_path)
 99        
100        # Make sure the file exists
101        if not refmasslist.exists():
102            raise FileExistsError("File does not exist: %s" % refmasslist)
103        
104        # Read all lines from the file (handling S3 vs local differently)
105        if isinstance(refmasslist, S3Path):
106            # For S3, read the file in binary, then decode to string and split into lines.
107            content = refmasslist.open("rb").read()
108            all_lines = content.decode("utf-8").splitlines(keepends=True)
109        else:
110            # For a local file, open in text mode and read lines.
111            with refmasslist.open("r") as f:
112                all_lines = f.readlines()
113        
114        # Identify the index of the first line of the actual data.
115        # We assume header lines start with '#' (or '##') and ignore blank lines.
116        data_start_idx = 0
117        for idx, line in enumerate(all_lines):
118            if line.strip() and not line.lstrip().startswith("#"):
119                data_start_idx = idx
120                break
121        
122        # If there are not enough lines to guess the dialect, throw an error
123        if data_start_idx >= len(all_lines):
124            raise ValueError("The file does not appear to contain any data lines.")
125        
126        # Use a couple of the data lines to let csv.Sniffer detect the delimiter
127        sample_lines = "".join(all_lines[data_start_idx:data_start_idx+2])
128        try:
129            dialect = csv.Sniffer().sniff(sample_lines)
130            delimiter = dialect.delimiter
131        except csv.Error:
132            # If csv.Sniffer fails, default to a common delimiter (e.g., comma)
133            delimiter = ","
134        
135        # Join the lines from the beginning of data (this might include a blank line) 
136        joined_data = "".join(all_lines[data_start_idx:])
137        
138        # Depending on whether the file is S3 or local, wrap the data as needed for pandas
139        if isinstance(refmasslist, S3Path):
140            data_buffer = BytesIO(joined_data.encode("utf-8"))
141        else:
142            data_buffer = StringIO(joined_data)
143        
144        # Read data into a DataFrame.
145        # Adjust columns and names as needed – here we assume at least 2 columns:
146        df_ref = pd.read_csv(data_buffer,
147                            sep=delimiter,
148                            header=None,
149                            usecols=[0, 1],   # Modify if more columns are required.
150                            names=["Formula", "m/z"])
151        
152        df_ref.sort_values(by="m/z", ascending=True, inplace=True)
153        
154        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
155            print("Reference mass list loaded - {} calibration masses loaded.".format(len(df_ref)))
156        
157        return df_ref
158
159    def gen_ref_mass_list_from_assigned(self, min_conf: float = 0.7):
160        """Generate reference mass list from assigned masses
161
162        This function will generate a ref mass dataframe object from an assigned corems mass spec obj
163        using assigned masses above a certain minimum confidence threshold.
164
165        This function needs to be retested and check it is covered in the unit tests.
166
167        Parameters
168        ----------
169        min_conf : float, optional
170            minimum confidence score. The default is 0.7.
171
172        Returns
173        -------
174        df_ref : Pandas DataFrame
175            reference mass list - based on calculated masses.
176
177        """
178        # TODO this function needs to be retested and check it is covered in the unit tests
179        df = self.mass_spectrum.to_dataframe()
180        df = df[df["Confidence Score"] > min_conf]
181        df_ref = pd.DataFrame(columns=["m/z"])
182        df_ref["m/z"] = df["Calculated m/z"]
183        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
184            print(
185                "Reference mass list generated - "
186                + str(len(df_ref))
187                + " calibration masses."
188            )
189        return df_ref
190
191    def find_calibration_points(
192        self,
193        df_ref,
194        calib_ppm_error_threshold: tuple[float, float] = (-1, 1),
195        calib_snr_threshold: float = 5,
196        calibration_ref_match_method: str = "merged",
197        calibration_ref_match_tolerance: float = 0.003,
198        calibration_ref_match_std_raw_error_limit: float = 1.5,
199    ):
200        """Function to find calibration points in the mass spectrum
201
202        Based on the reference mass list.
203
204        Parameters
205        ----------
206        df_ref : Pandas DataFrame
207            reference mass list for recalibration.
208        calib_ppm_error_threshold : tuple of floats, optional
209            ppm error for finding calibration masses in the spectrum. The default is -1,1.
210            Note: This is based on the calculation of ppm = ((mz_measure - mz_theoretical)/mz_theoretical)*1e6.
211                Some software does this the other way around and value signs must be inverted for that to work.
212        calib_snr_threshold : float, optional
213            snr threshold for finding calibration masses in the spectrum. The default is 5.
214            If SNR data is unavailable, peaks are filtered by intensity percentile using the formula:
215            percentile = max(5, 100 - calib_snr_threshold)
216        calibration_ref_match_method : str, optional
217            method for matching calibration references. The default is "merged".
218        calibration_ref_match_tolerance : float, optional
219            tolerance for matching calibration references. The default is 0.003.
220        calibration_ref_match_std_raw_error_limit : float, optional
221            standard deviation raw error limit for calibration references. The default is 1.5.
222
223        Returns
224        -------
225        cal_peaks_mz : list of floats
226            masses of measured ions to use in calibration routine
227        cal_refs_mz : list of floats
228            reference mz values of found calibration points.
229
230        """
231
232        # Check if SNR data is available by testing the first peak
233        use_snr = False
234        if len(self.mass_spectrum.mspeaks) > 0:
235            first_peak = self.mass_spectrum.mspeaks[0]
236            if (hasattr(first_peak, 'signal_to_noise') and 
237                first_peak.signal_to_noise is not None and 
238                not np.isnan(first_peak.signal_to_noise) and
239                first_peak.signal_to_noise > 0):
240                use_snr = True
241
242        # This approach is much more efficient and expedient than the original implementation.
243        peaks_mz = []
244        peaks_intensity = []
245        
246        if use_snr:
247            # Use SNR filtering
248            for x in self.mass_spectrum.mspeaks:
249                if x.signal_to_noise > calib_snr_threshold:
250                    if self.mzsegment:
251                        if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
252                            peaks_mz.append(x.mz_exp)
253                    else:
254                        peaks_mz.append(x.mz_exp)
255        else:
256            # Fallback to intensity percentile filtering
257            intensity_percentile = max(5, 100 - calib_snr_threshold)
258            warnings.warn(
259                f"SNR data unavailable for calibration. Using intensity-based filtering instead. "
260                f"SNR threshold of {calib_snr_threshold} corresponds to intensity percentile >= {intensity_percentile}%."
261            )
262            
263            # Collect all peaks and their intensities
264            all_peaks_data = []
265            for x in self.mass_spectrum.mspeaks:
266                if self.mzsegment:
267                    if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
268                        all_peaks_data.append((x.mz_exp, x.abundance))
269                else:
270                    all_peaks_data.append((x.mz_exp, x.abundance))
271            
272            if all_peaks_data:
273                peaks_mz_list, intensities = zip(*all_peaks_data)
274                intensity_threshold = np.percentile(intensities, intensity_percentile)
275                
276                for mz, intensity in all_peaks_data:
277                    if intensity >= intensity_threshold:
278                        peaks_mz.append(mz)
279        
280        peaks_mz = np.asarray(peaks_mz)
281
282        if calibration_ref_match_method == "legacy":
283            # This legacy approach iterates through each reference match and finds the entries within 1 mz and within the user defined PPM error threshold
284            # Then it removes ambiguities - which means the calibration threshold hasto be very tight.
285            cal_peaks_mz = []
286            cal_refs_mz = []
287            for mzref in df_ref["m/z"]:
288                tmp_peaks_mz = peaks_mz[abs(peaks_mz - mzref) < 1]
289                for mzmeas in tmp_peaks_mz:
290                    delta_mass = ((mzmeas - mzref) / mzref) * 1e6
291                    if delta_mass < max(calib_ppm_error_threshold):
292                        if delta_mass > min(calib_ppm_error_threshold):
293                            cal_peaks_mz.append(mzmeas)
294                            cal_refs_mz.append(mzref)
295
296            # To remove entries with duplicated indices (reference masses matching multiple peaks)
297            tmpdf = pd.Series(index=cal_refs_mz, data=cal_peaks_mz, dtype=float)
298            tmpdf = tmpdf[~tmpdf.index.duplicated(keep=False)]
299
300            cal_peaks_mz = list(tmpdf.values)
301            cal_refs_mz = list(tmpdf.index)
302        elif calibration_ref_match_method == "merged":
303            # This is a new approach (August 2024) which uses Pandas 'merged_asof' to find the peaks closest in m/z between
304            # reference and measured masses. This is a quicker way to match, and seems to get more matches.
305            # It may not work as well when the data are far from correc initial mass
306            # e.g. if the correct peak is further from the reference than an incorrect peak.
307            meas_df = pd.DataFrame(columns=["meas_m/z"], data=peaks_mz)
308            tolerance = calibration_ref_match_tolerance
309            merged_df = pd.merge_asof(
310                df_ref,
311                meas_df,
312                left_on="m/z",
313                right_on="meas_m/z",
314                tolerance=tolerance,
315                direction="nearest",
316            )
317            merged_df.dropna(how="any", inplace=True)
318            merged_df["Error_ppm"] = (
319                (merged_df["meas_m/z"] - merged_df["m/z"]) / merged_df["m/z"]
320            ) * 1e6
321            median_raw_error = merged_df["Error_ppm"].median()
322            std_raw_error = merged_df["Error_ppm"].std()
323            if std_raw_error > calibration_ref_match_std_raw_error_limit:
324                std_raw_error = calibration_ref_match_std_raw_error_limit
325            self.mass_spectrum.calibration_raw_error_median = median_raw_error
326            self.mass_spectrum.calibration_raw_error_stdev = std_raw_error
327            merged_df = merged_df[
328                (merged_df["Error_ppm"] > (median_raw_error - 1.5 * std_raw_error))
329                & (merged_df["Error_ppm"] < (median_raw_error + 1.5 * std_raw_error))
330            ]
331            # merged_df= merged_df[(merged_df['Error_ppm']>min(calib_ppm_error_threshold))&(merged_df['Error_ppm']<max(calib_ppm_error_threshold))]
332            cal_peaks_mz = list(merged_df["meas_m/z"])
333            cal_refs_mz = list(merged_df["m/z"])
334        else:
335            raise ValueError(f"{calibration_ref_match_method} not allowed.")
336
337        # it is crucial the mass lists are in same order
338        # corems likes to do masses from high to low.
339        cal_refs_mz.sort(reverse=False)
340        cal_peaks_mz.sort(reverse=False)
341        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
342            print(
343                str(len(cal_peaks_mz))
344                + " calibration points matched within thresholds."
345            )
346        return cal_peaks_mz, cal_refs_mz
347
348    def robust_calib(
349        self,
350        param: list[float],
351        cal_peaks_mz: list[float],
352        cal_refs_mz: list[float],
353        order: int = 1,
354    ):
355        """Recalibration function
356
357        Computes the rms of m/z errors to minimize when calibrating.
358        This is adapted from from spike.
359
360        Parameters
361        ----------
362        param : list of floats
363            generated by minimize function from scipy optimize.
364        cal_peaks_mz : list of floats
365            masses of measured peaks to use in mass calibration.
366        cal_peaks_mz : list of floats
367            reference mz values of found calibration points.
368        order : int, optional
369            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
370
371        Returns
372        -------
373        rmserror : float
374            root mean square mass error for calibration points.
375
376        """
377        Aterm = param[0]
378        Bterm = param[1]
379        try:
380            Cterm = param[2]
381        except IndexError:
382            pass
383
384        # get the mspeaks from the mass spectrum object which were calibration points
385        # mspeaks = [self.mass_spectrum.mspeaks[x] for x in imzmeas]
386        # get their calibrated mass values
387        # mspeakmzs = [x.mz_cal for x in mspeaks]
388        cal_peaks_mz = np.asarray(cal_peaks_mz)
389
390        # linearz
391        if order == 1:
392            ref_recal_points = (Aterm * cal_peaks_mz) + Bterm
393        # quadratic
394        elif order == 2:
395            ref_recal_points = (Aterm * (cal_peaks_mz)) + (
396                Bterm * np.power((cal_peaks_mz), 2) + Cterm
397            )
398
399        # sort both the calibration points (measured, recalibrated)
400        ref_recal_points.sort()
401        # and sort the calibration points (theoretical, predefined)
402        cal_refs_mz.sort()
403
404        # calculate the ppm error for each calibration point
405        error = ((ref_recal_points - cal_refs_mz) / cal_refs_mz) * 1e6
406        # calculate the root mean square error - this is our target to minimize
407        rmserror = np.sqrt(np.mean(error**2))
408        return rmserror
409
410    def recalibrate_mass_spectrum(
411        self,
412        cal_peaks_mz: list[float],
413        cal_refs_mz: list[float],
414        order: int = 1,
415        diagnostic: bool = False,
416    ):
417        """Main recalibration function which uses a robust linear regression
418
419        This function performs the recalibration of the mass spectrum object.
420        It iteratively applies
421
422        Parameters
423        ----------
424        cal_peaks_mz : list of float
425            masses of measured peaks to use in mass calibration.
426        cal_refs_mz : list of float
427            reference mz values of found calibration points.
428        order : int, optional
429            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
430
431        Returns
432        -------
433        mass_spectrum : CoreMS mass spectrum object
434            Calibrated mass spectrum object
435
436
437        Notes
438        -----
439        This function is adapted, in part, from the SPIKE project [1,2] and is based on the robust linear regression method.
440
441        References
442        ----------
443        1.  Chiron L., Coutouly M-A., Starck J-P., Rolando C., Delsuc M-A.
444            SPIKE a Processing Software dedicated to Fourier Spectroscopies
445            https://arxiv.org/abs/1608.06777 (2016)
446        2.  SPIKE - https://github.com/spike-project/spike
447
448        """
449        # initialise parameters for recalibration
450        # these are the 'Aterm, Bterm, Cterm'
451        # as spectra are already freq->mz calibrated, these terms are very small
452        # may be beneficial to formally separate them from the freq->mz terms
453        if order == 1:
454            Po = [1, 0]
455        elif order == 2:
456            Po = [1, 0, 0]
457
458        if len(cal_peaks_mz) >= 2:
459            if self.mzsegment:  # If only part of the spectrum is to be recalibrated
460                mz_exp_peaks = np.array(
461                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
462                )
463                # Split the array into two parts - one to recailbrate, one to keep unchanged.
464                mz_exp_peaks_tocal = mz_exp_peaks[
465                    (mz_exp_peaks >= min(self.mzsegment))
466                    & (mz_exp_peaks <= max(self.mzsegment))
467                ]
468                mz_exp_peaks_unchanged = mz_exp_peaks[
469                    ~(mz_exp_peaks >= min(self.mzsegment))
470                    | ~(mz_exp_peaks <= max(self.mzsegment))
471                ]
472                # TODO: - segmented calibration needs a way to better track the calibration args/values...
473                if not self.mass_spectrum.is_centroid:
474                    mz_exp_profile = np.array(self.mass_spectrum.mz_exp_profile)
475                    # Split the array into two parts - one to recailbrate, one to keep unchanged.
476                    mz_exp_profile_tocal = mz_exp_profile[
477                        (mz_exp_profile >= min(self.mzsegment))
478                        & (mz_exp_profile <= max(self.mzsegment))
479                    ]
480                    mz_exp_profile_unchanged = mz_exp_profile[
481                        ~(mz_exp_profile >= min(self.mzsegment))
482                        | ~(mz_exp_profile <= max(self.mzsegment))
483                    ]
484            else:  # if just recalibrating the whole spectrum
485                mz_exp_peaks_tocal = np.array(
486                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
487                )
488                if not self.mass_spectrum.is_centroid:
489                    mz_exp_profile_tocal = np.array(self.mass_spectrum.mz_exp_profile)
490
491            minimize_method = self.mass_spectrum.settings.calib_minimize_method
492            res = minimize(
493                self.robust_calib,
494                Po,
495                args=(cal_peaks_mz, cal_refs_mz, order),
496                method=minimize_method,
497            )
498            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
499                print(
500                    "minimize function completed with RMS error of: {:0.3f} ppm".format(
501                        res["fun"]
502                    )
503                )
504                print(
505                    "minimize function performed {:1d} fn evals and {:1d} iterations".format(
506                        res["nfev"], res["nit"]
507                    )
508                )
509            Pn = res.x
510
511            # mz_exp_ms = np.array([mspeak.mz_exp for mspeak in self.mass_spectrum])
512
513            if order == 1:
514                mz_domain = (Pn[0] * mz_exp_peaks_tocal) + Pn[1]
515                if not self.mass_spectrum.is_centroid:
516                    mz_profile_calc = (Pn[0] * mz_exp_profile_tocal) + Pn[1]
517
518            elif order == 2:
519                mz_domain = (Pn[0] * (mz_exp_peaks_tocal)) + (
520                    Pn[1] * np.power((mz_exp_peaks_tocal), 2) + Pn[2]
521                )
522
523                if not self.mass_spectrum.is_centroid:
524                    mz_profile_calc = (Pn[0] * (mz_exp_profile_tocal)) + (
525                        Pn[1] * np.power((mz_exp_profile_tocal), 2) + Pn[2]
526                    )
527
528            if self.mzsegment:
529                # Recombine the mass domains
530                mz_domain = np.concatenate([mz_domain, mz_exp_peaks_unchanged])
531                mz_domain.sort()
532                if not self.mass_spectrum.is_centroid:
533                    mz_profile_calc = np.concatenate(
534                        [mz_profile_calc, mz_exp_profile_unchanged]
535                    )
536                    mz_profile_calc.sort()
537                # Sort them
538                if (
539                    mz_exp_peaks[0] > mz_exp_peaks[1]
540                ):  # If originally descending mass order
541                    mz_domain = mz_domain[::-1]
542                    if not self.mass_spectrum.is_centroid:
543                        mz_profile_calc = mz_profile_calc[::-1]
544
545            self.mass_spectrum.mz_cal = mz_domain
546            if not self.mass_spectrum.is_centroid:
547                self.mass_spectrum.mz_cal_profile = mz_profile_calc
548
549            self.mass_spectrum.calibration_order = order
550            self.mass_spectrum.calibration_RMS = float(res["fun"])
551            self.mass_spectrum.calibration_points = int(len(cal_refs_mz))
552            self.mass_spectrum.calibration_ref_mzs = cal_refs_mz
553            self.mass_spectrum.calibration_meas_mzs = cal_peaks_mz
554
555            self.mass_spectrum.calibration_segment = self.mzsegment
556
557            if diagnostic:
558                return self.mass_spectrum, res
559            return self.mass_spectrum
560        else:
561            warnings.warn("Too few calibration points - aborting.")
562            return self.mass_spectrum
563
564    def run(self):
565        """Run the calibration routine
566
567        This function runs the calibration routine.
568
569        """
570        calib_snr_threshold = self.mass_spectrum.settings.calib_sn_threshold
571        max_calib_ppm_error = self.mass_spectrum.settings.max_calib_ppm_error
572        min_calib_ppm_error = self.mass_spectrum.settings.min_calib_ppm_error
573        calib_pol_order = self.mass_spectrum.settings.calib_pol_order
574        calibration_ref_match_method = (
575            self.mass_spectrum.settings.calibration_ref_match_method
576        )
577        calibration_ref_match_tolerance = (
578            self.mass_spectrum.settings.calibration_ref_match_tolerance
579        )
580        calibration_ref_match_std_raw_error_limit = (
581            self.mass_spectrum.settings.calibration_ref_match_std_raw_error_limit
582        )
583
584        # load reference mass list
585        df_ref = self.load_ref_mass_list()
586
587        # find calibration points
588        cal_peaks_mz, cal_refs_mz = self.find_calibration_points(
589            df_ref,
590            calib_ppm_error_threshold=(min_calib_ppm_error, max_calib_ppm_error),
591            calib_snr_threshold=calib_snr_threshold,
592            calibration_ref_match_method=calibration_ref_match_method,
593            calibration_ref_match_tolerance=calibration_ref_match_tolerance,
594            calibration_ref_match_std_raw_error_limit=calibration_ref_match_std_raw_error_limit,
595        )
596        if len(cal_peaks_mz) == 2:
597            self.mass_spectrum.settings.calib_pol_order = 1
598            calib_pol_order = 1
599            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
600                print("Only 2 calibration points found, forcing a linear recalibration")
601        elif len(cal_peaks_mz) < 2:
602            warnings.warn("Too few calibration points found, function will fail")
603        self.recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=calib_pol_order)

MzDomainCalibration class for recalibrating mass spectra

Parameters
  • mass_spectrum (CoreMS MassSpectrum Object): The mass spectrum to be calibrated.
  • ref_masslist (str): The path to a reference mass list.
  • mzsegment (tuple of floats, optional): The mz range to recalibrate, or None. Used for calibration of specific parts of the mz domain at a time. Future work - allow multiple mzsegments to be passed.
Attributes
  • mass_spectrum (CoreMS MassSpectrum Object): The mass spectrum to be calibrated.
  • mzsegment (tuple of floats or None): The mz range to recalibrate, or None.
  • ref_mass_list_path (str or Path): The path to the reference mass list.
Methods
  • run(). Main function to run this class.
  • load_ref_mass_list(). Load reference mass list (Bruker format).
  • gen_ref_mass_list_from_assigned(min_conf=0.7). Generate reference mass list from assigned masses.
  • find_calibration_points(df_ref, calib_ppm_error_threshold=(-1, 1), calib_snr_threshold=5). Find calibration points in the mass spectrum based on the reference mass list.
  • robust_calib(param, cal_peaks_mz, cal_refs_mz, order=1). Recalibration function.
  • recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=1, diagnostic=False). Main recalibration function which uses a robust linear regression.
MzDomainCalibration(mass_spectrum, ref_masslist, mzsegment=None)
63    def __init__(self, mass_spectrum, ref_masslist, mzsegment=None):
64        self.mass_spectrum = mass_spectrum
65        self.mzsegment = mzsegment
66
67        # define reference mass list - bruker .ref format
68        self.ref_mass_list_path = ref_masslist
69        if self.mass_spectrum.percentile_assigned(mute_output=True)[0] != 0:
70            warnings.warn(
71                "Warning: calibrating spectra which have already been assigned may yield erroneous results"
72            )
73        self.mass_spectrum.mz_cal = self.mass_spectrum.mz_exp
74        self.mass_spectrum.mz_cal_profile = self.mass_spectrum._mz_exp
75
76        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
77            print(
78                "MS Obj loaded - " + str(len(mass_spectrum.mspeaks)) + " peaks found."
79            )
mass_spectrum
mzsegment
ref_mass_list_path
def load_ref_mass_list(self):
 81    def load_ref_mass_list(self):
 82        """
 83        Load reference mass list (Bruker format).
 84
 85        Loads in a reference mass list from a .ref file. Some versions of
 86        Bruker's software produce .ref files with a different format where
 87        the header lines (starting with '#' or '##') and delimiters may vary.
 88        The file may be located locally or on S3 and will be handled accordingly.
 89        
 90        Returns
 91        -------
 92        df_ref : Pandas DataFrame
 93            Reference mass list object.
 94        """
 95        # Get a Path-like object from the input path string or S3Path
 96        refmasslist = (Path(self.ref_mass_list_path)
 97                    if isinstance(self.ref_mass_list_path, str)
 98                    else self.ref_mass_list_path)
 99        
100        # Make sure the file exists
101        if not refmasslist.exists():
102            raise FileExistsError("File does not exist: %s" % refmasslist)
103        
104        # Read all lines from the file (handling S3 vs local differently)
105        if isinstance(refmasslist, S3Path):
106            # For S3, read the file in binary, then decode to string and split into lines.
107            content = refmasslist.open("rb").read()
108            all_lines = content.decode("utf-8").splitlines(keepends=True)
109        else:
110            # For a local file, open in text mode and read lines.
111            with refmasslist.open("r") as f:
112                all_lines = f.readlines()
113        
114        # Identify the index of the first line of the actual data.
115        # We assume header lines start with '#' (or '##') and ignore blank lines.
116        data_start_idx = 0
117        for idx, line in enumerate(all_lines):
118            if line.strip() and not line.lstrip().startswith("#"):
119                data_start_idx = idx
120                break
121        
122        # If there are not enough lines to guess the dialect, throw an error
123        if data_start_idx >= len(all_lines):
124            raise ValueError("The file does not appear to contain any data lines.")
125        
126        # Use a couple of the data lines to let csv.Sniffer detect the delimiter
127        sample_lines = "".join(all_lines[data_start_idx:data_start_idx+2])
128        try:
129            dialect = csv.Sniffer().sniff(sample_lines)
130            delimiter = dialect.delimiter
131        except csv.Error:
132            # If csv.Sniffer fails, default to a common delimiter (e.g., comma)
133            delimiter = ","
134        
135        # Join the lines from the beginning of data (this might include a blank line) 
136        joined_data = "".join(all_lines[data_start_idx:])
137        
138        # Depending on whether the file is S3 or local, wrap the data as needed for pandas
139        if isinstance(refmasslist, S3Path):
140            data_buffer = BytesIO(joined_data.encode("utf-8"))
141        else:
142            data_buffer = StringIO(joined_data)
143        
144        # Read data into a DataFrame.
145        # Adjust columns and names as needed – here we assume at least 2 columns:
146        df_ref = pd.read_csv(data_buffer,
147                            sep=delimiter,
148                            header=None,
149                            usecols=[0, 1],   # Modify if more columns are required.
150                            names=["Formula", "m/z"])
151        
152        df_ref.sort_values(by="m/z", ascending=True, inplace=True)
153        
154        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
155            print("Reference mass list loaded - {} calibration masses loaded.".format(len(df_ref)))
156        
157        return df_ref

Load reference mass list (Bruker format).

Loads in a reference mass list from a .ref file. Some versions of Bruker's software produce .ref files with a different format where the header lines (starting with '#' or '##') and delimiters may vary. The file may be located locally or on S3 and will be handled accordingly.

Returns
  • df_ref (Pandas DataFrame): Reference mass list object.
def gen_ref_mass_list_from_assigned(self, min_conf: float = 0.7):
159    def gen_ref_mass_list_from_assigned(self, min_conf: float = 0.7):
160        """Generate reference mass list from assigned masses
161
162        This function will generate a ref mass dataframe object from an assigned corems mass spec obj
163        using assigned masses above a certain minimum confidence threshold.
164
165        This function needs to be retested and check it is covered in the unit tests.
166
167        Parameters
168        ----------
169        min_conf : float, optional
170            minimum confidence score. The default is 0.7.
171
172        Returns
173        -------
174        df_ref : Pandas DataFrame
175            reference mass list - based on calculated masses.
176
177        """
178        # TODO this function needs to be retested and check it is covered in the unit tests
179        df = self.mass_spectrum.to_dataframe()
180        df = df[df["Confidence Score"] > min_conf]
181        df_ref = pd.DataFrame(columns=["m/z"])
182        df_ref["m/z"] = df["Calculated m/z"]
183        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
184            print(
185                "Reference mass list generated - "
186                + str(len(df_ref))
187                + " calibration masses."
188            )
189        return df_ref

Generate reference mass list from assigned masses

This function will generate a ref mass dataframe object from an assigned corems mass spec obj using assigned masses above a certain minimum confidence threshold.

This function needs to be retested and check it is covered in the unit tests.

Parameters
  • min_conf (float, optional): minimum confidence score. The default is 0.7.
Returns
  • df_ref (Pandas DataFrame): reference mass list - based on calculated masses.
def find_calibration_points( self, df_ref, calib_ppm_error_threshold: tuple[float, float] = (-1, 1), calib_snr_threshold: float = 5, calibration_ref_match_method: str = 'merged', calibration_ref_match_tolerance: float = 0.003, calibration_ref_match_std_raw_error_limit: float = 1.5):
191    def find_calibration_points(
192        self,
193        df_ref,
194        calib_ppm_error_threshold: tuple[float, float] = (-1, 1),
195        calib_snr_threshold: float = 5,
196        calibration_ref_match_method: str = "merged",
197        calibration_ref_match_tolerance: float = 0.003,
198        calibration_ref_match_std_raw_error_limit: float = 1.5,
199    ):
200        """Function to find calibration points in the mass spectrum
201
202        Based on the reference mass list.
203
204        Parameters
205        ----------
206        df_ref : Pandas DataFrame
207            reference mass list for recalibration.
208        calib_ppm_error_threshold : tuple of floats, optional
209            ppm error for finding calibration masses in the spectrum. The default is -1,1.
210            Note: This is based on the calculation of ppm = ((mz_measure - mz_theoretical)/mz_theoretical)*1e6.
211                Some software does this the other way around and value signs must be inverted for that to work.
212        calib_snr_threshold : float, optional
213            snr threshold for finding calibration masses in the spectrum. The default is 5.
214            If SNR data is unavailable, peaks are filtered by intensity percentile using the formula:
215            percentile = max(5, 100 - calib_snr_threshold)
216        calibration_ref_match_method : str, optional
217            method for matching calibration references. The default is "merged".
218        calibration_ref_match_tolerance : float, optional
219            tolerance for matching calibration references. The default is 0.003.
220        calibration_ref_match_std_raw_error_limit : float, optional
221            standard deviation raw error limit for calibration references. The default is 1.5.
222
223        Returns
224        -------
225        cal_peaks_mz : list of floats
226            masses of measured ions to use in calibration routine
227        cal_refs_mz : list of floats
228            reference mz values of found calibration points.
229
230        """
231
232        # Check if SNR data is available by testing the first peak
233        use_snr = False
234        if len(self.mass_spectrum.mspeaks) > 0:
235            first_peak = self.mass_spectrum.mspeaks[0]
236            if (hasattr(first_peak, 'signal_to_noise') and 
237                first_peak.signal_to_noise is not None and 
238                not np.isnan(first_peak.signal_to_noise) and
239                first_peak.signal_to_noise > 0):
240                use_snr = True
241
242        # This approach is much more efficient and expedient than the original implementation.
243        peaks_mz = []
244        peaks_intensity = []
245        
246        if use_snr:
247            # Use SNR filtering
248            for x in self.mass_spectrum.mspeaks:
249                if x.signal_to_noise > calib_snr_threshold:
250                    if self.mzsegment:
251                        if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
252                            peaks_mz.append(x.mz_exp)
253                    else:
254                        peaks_mz.append(x.mz_exp)
255        else:
256            # Fallback to intensity percentile filtering
257            intensity_percentile = max(5, 100 - calib_snr_threshold)
258            warnings.warn(
259                f"SNR data unavailable for calibration. Using intensity-based filtering instead. "
260                f"SNR threshold of {calib_snr_threshold} corresponds to intensity percentile >= {intensity_percentile}%."
261            )
262            
263            # Collect all peaks and their intensities
264            all_peaks_data = []
265            for x in self.mass_spectrum.mspeaks:
266                if self.mzsegment:
267                    if min(self.mzsegment) <= x.mz_exp <= max(self.mzsegment):
268                        all_peaks_data.append((x.mz_exp, x.abundance))
269                else:
270                    all_peaks_data.append((x.mz_exp, x.abundance))
271            
272            if all_peaks_data:
273                peaks_mz_list, intensities = zip(*all_peaks_data)
274                intensity_threshold = np.percentile(intensities, intensity_percentile)
275                
276                for mz, intensity in all_peaks_data:
277                    if intensity >= intensity_threshold:
278                        peaks_mz.append(mz)
279        
280        peaks_mz = np.asarray(peaks_mz)
281
282        if calibration_ref_match_method == "legacy":
283            # This legacy approach iterates through each reference match and finds the entries within 1 mz and within the user defined PPM error threshold
284            # Then it removes ambiguities - which means the calibration threshold hasto be very tight.
285            cal_peaks_mz = []
286            cal_refs_mz = []
287            for mzref in df_ref["m/z"]:
288                tmp_peaks_mz = peaks_mz[abs(peaks_mz - mzref) < 1]
289                for mzmeas in tmp_peaks_mz:
290                    delta_mass = ((mzmeas - mzref) / mzref) * 1e6
291                    if delta_mass < max(calib_ppm_error_threshold):
292                        if delta_mass > min(calib_ppm_error_threshold):
293                            cal_peaks_mz.append(mzmeas)
294                            cal_refs_mz.append(mzref)
295
296            # To remove entries with duplicated indices (reference masses matching multiple peaks)
297            tmpdf = pd.Series(index=cal_refs_mz, data=cal_peaks_mz, dtype=float)
298            tmpdf = tmpdf[~tmpdf.index.duplicated(keep=False)]
299
300            cal_peaks_mz = list(tmpdf.values)
301            cal_refs_mz = list(tmpdf.index)
302        elif calibration_ref_match_method == "merged":
303            # This is a new approach (August 2024) which uses Pandas 'merged_asof' to find the peaks closest in m/z between
304            # reference and measured masses. This is a quicker way to match, and seems to get more matches.
305            # It may not work as well when the data are far from correc initial mass
306            # e.g. if the correct peak is further from the reference than an incorrect peak.
307            meas_df = pd.DataFrame(columns=["meas_m/z"], data=peaks_mz)
308            tolerance = calibration_ref_match_tolerance
309            merged_df = pd.merge_asof(
310                df_ref,
311                meas_df,
312                left_on="m/z",
313                right_on="meas_m/z",
314                tolerance=tolerance,
315                direction="nearest",
316            )
317            merged_df.dropna(how="any", inplace=True)
318            merged_df["Error_ppm"] = (
319                (merged_df["meas_m/z"] - merged_df["m/z"]) / merged_df["m/z"]
320            ) * 1e6
321            median_raw_error = merged_df["Error_ppm"].median()
322            std_raw_error = merged_df["Error_ppm"].std()
323            if std_raw_error > calibration_ref_match_std_raw_error_limit:
324                std_raw_error = calibration_ref_match_std_raw_error_limit
325            self.mass_spectrum.calibration_raw_error_median = median_raw_error
326            self.mass_spectrum.calibration_raw_error_stdev = std_raw_error
327            merged_df = merged_df[
328                (merged_df["Error_ppm"] > (median_raw_error - 1.5 * std_raw_error))
329                & (merged_df["Error_ppm"] < (median_raw_error + 1.5 * std_raw_error))
330            ]
331            # merged_df= merged_df[(merged_df['Error_ppm']>min(calib_ppm_error_threshold))&(merged_df['Error_ppm']<max(calib_ppm_error_threshold))]
332            cal_peaks_mz = list(merged_df["meas_m/z"])
333            cal_refs_mz = list(merged_df["m/z"])
334        else:
335            raise ValueError(f"{calibration_ref_match_method} not allowed.")
336
337        # it is crucial the mass lists are in same order
338        # corems likes to do masses from high to low.
339        cal_refs_mz.sort(reverse=False)
340        cal_peaks_mz.sort(reverse=False)
341        if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
342            print(
343                str(len(cal_peaks_mz))
344                + " calibration points matched within thresholds."
345            )
346        return cal_peaks_mz, cal_refs_mz

Function to find calibration points in the mass spectrum

Based on the reference mass list.

Parameters
  • df_ref (Pandas DataFrame): reference mass list for recalibration.
  • calib_ppm_error_threshold (tuple of floats, optional): ppm error for finding calibration masses in the spectrum. The default is -1,1. Note: This is based on the calculation of ppm = ((mz_measure - mz_theoretical)/mz_theoretical)*1e6. Some software does this the other way around and value signs must be inverted for that to work.
  • calib_snr_threshold (float, optional): snr threshold for finding calibration masses in the spectrum. The default is 5. If SNR data is unavailable, peaks are filtered by intensity percentile using the formula: percentile = max(5, 100 - calib_snr_threshold)
  • calibration_ref_match_method (str, optional): method for matching calibration references. The default is "merged".
  • calibration_ref_match_tolerance (float, optional): tolerance for matching calibration references. The default is 0.003.
  • calibration_ref_match_std_raw_error_limit (float, optional): standard deviation raw error limit for calibration references. The default is 1.5.
Returns
  • cal_peaks_mz (list of floats): masses of measured ions to use in calibration routine
  • cal_refs_mz (list of floats): reference mz values of found calibration points.
def robust_calib( self, param: list[float], cal_peaks_mz: list[float], cal_refs_mz: list[float], order: int = 1):
348    def robust_calib(
349        self,
350        param: list[float],
351        cal_peaks_mz: list[float],
352        cal_refs_mz: list[float],
353        order: int = 1,
354    ):
355        """Recalibration function
356
357        Computes the rms of m/z errors to minimize when calibrating.
358        This is adapted from from spike.
359
360        Parameters
361        ----------
362        param : list of floats
363            generated by minimize function from scipy optimize.
364        cal_peaks_mz : list of floats
365            masses of measured peaks to use in mass calibration.
366        cal_peaks_mz : list of floats
367            reference mz values of found calibration points.
368        order : int, optional
369            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
370
371        Returns
372        -------
373        rmserror : float
374            root mean square mass error for calibration points.
375
376        """
377        Aterm = param[0]
378        Bterm = param[1]
379        try:
380            Cterm = param[2]
381        except IndexError:
382            pass
383
384        # get the mspeaks from the mass spectrum object which were calibration points
385        # mspeaks = [self.mass_spectrum.mspeaks[x] for x in imzmeas]
386        # get their calibrated mass values
387        # mspeakmzs = [x.mz_cal for x in mspeaks]
388        cal_peaks_mz = np.asarray(cal_peaks_mz)
389
390        # linearz
391        if order == 1:
392            ref_recal_points = (Aterm * cal_peaks_mz) + Bterm
393        # quadratic
394        elif order == 2:
395            ref_recal_points = (Aterm * (cal_peaks_mz)) + (
396                Bterm * np.power((cal_peaks_mz), 2) + Cterm
397            )
398
399        # sort both the calibration points (measured, recalibrated)
400        ref_recal_points.sort()
401        # and sort the calibration points (theoretical, predefined)
402        cal_refs_mz.sort()
403
404        # calculate the ppm error for each calibration point
405        error = ((ref_recal_points - cal_refs_mz) / cal_refs_mz) * 1e6
406        # calculate the root mean square error - this is our target to minimize
407        rmserror = np.sqrt(np.mean(error**2))
408        return rmserror

Recalibration function

Computes the rms of m/z errors to minimize when calibrating. This is adapted from from spike.

Parameters
  • param (list of floats): generated by minimize function from scipy optimize.
  • cal_peaks_mz (list of floats): masses of measured peaks to use in mass calibration.
  • cal_peaks_mz (list of floats): reference mz values of found calibration points.
  • order (int, optional): order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
Returns
  • rmserror (float): root mean square mass error for calibration points.
def recalibrate_mass_spectrum( self, cal_peaks_mz: list[float], cal_refs_mz: list[float], order: int = 1, diagnostic: bool = False):
410    def recalibrate_mass_spectrum(
411        self,
412        cal_peaks_mz: list[float],
413        cal_refs_mz: list[float],
414        order: int = 1,
415        diagnostic: bool = False,
416    ):
417        """Main recalibration function which uses a robust linear regression
418
419        This function performs the recalibration of the mass spectrum object.
420        It iteratively applies
421
422        Parameters
423        ----------
424        cal_peaks_mz : list of float
425            masses of measured peaks to use in mass calibration.
426        cal_refs_mz : list of float
427            reference mz values of found calibration points.
428        order : int, optional
429            order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
430
431        Returns
432        -------
433        mass_spectrum : CoreMS mass spectrum object
434            Calibrated mass spectrum object
435
436
437        Notes
438        -----
439        This function is adapted, in part, from the SPIKE project [1,2] and is based on the robust linear regression method.
440
441        References
442        ----------
443        1.  Chiron L., Coutouly M-A., Starck J-P., Rolando C., Delsuc M-A.
444            SPIKE a Processing Software dedicated to Fourier Spectroscopies
445            https://arxiv.org/abs/1608.06777 (2016)
446        2.  SPIKE - https://github.com/spike-project/spike
447
448        """
449        # initialise parameters for recalibration
450        # these are the 'Aterm, Bterm, Cterm'
451        # as spectra are already freq->mz calibrated, these terms are very small
452        # may be beneficial to formally separate them from the freq->mz terms
453        if order == 1:
454            Po = [1, 0]
455        elif order == 2:
456            Po = [1, 0, 0]
457
458        if len(cal_peaks_mz) >= 2:
459            if self.mzsegment:  # If only part of the spectrum is to be recalibrated
460                mz_exp_peaks = np.array(
461                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
462                )
463                # Split the array into two parts - one to recailbrate, one to keep unchanged.
464                mz_exp_peaks_tocal = mz_exp_peaks[
465                    (mz_exp_peaks >= min(self.mzsegment))
466                    & (mz_exp_peaks <= max(self.mzsegment))
467                ]
468                mz_exp_peaks_unchanged = mz_exp_peaks[
469                    ~(mz_exp_peaks >= min(self.mzsegment))
470                    | ~(mz_exp_peaks <= max(self.mzsegment))
471                ]
472                # TODO: - segmented calibration needs a way to better track the calibration args/values...
473                if not self.mass_spectrum.is_centroid:
474                    mz_exp_profile = np.array(self.mass_spectrum.mz_exp_profile)
475                    # Split the array into two parts - one to recailbrate, one to keep unchanged.
476                    mz_exp_profile_tocal = mz_exp_profile[
477                        (mz_exp_profile >= min(self.mzsegment))
478                        & (mz_exp_profile <= max(self.mzsegment))
479                    ]
480                    mz_exp_profile_unchanged = mz_exp_profile[
481                        ~(mz_exp_profile >= min(self.mzsegment))
482                        | ~(mz_exp_profile <= max(self.mzsegment))
483                    ]
484            else:  # if just recalibrating the whole spectrum
485                mz_exp_peaks_tocal = np.array(
486                    [mspeak.mz_exp for mspeak in self.mass_spectrum]
487                )
488                if not self.mass_spectrum.is_centroid:
489                    mz_exp_profile_tocal = np.array(self.mass_spectrum.mz_exp_profile)
490
491            minimize_method = self.mass_spectrum.settings.calib_minimize_method
492            res = minimize(
493                self.robust_calib,
494                Po,
495                args=(cal_peaks_mz, cal_refs_mz, order),
496                method=minimize_method,
497            )
498            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
499                print(
500                    "minimize function completed with RMS error of: {:0.3f} ppm".format(
501                        res["fun"]
502                    )
503                )
504                print(
505                    "minimize function performed {:1d} fn evals and {:1d} iterations".format(
506                        res["nfev"], res["nit"]
507                    )
508                )
509            Pn = res.x
510
511            # mz_exp_ms = np.array([mspeak.mz_exp for mspeak in self.mass_spectrum])
512
513            if order == 1:
514                mz_domain = (Pn[0] * mz_exp_peaks_tocal) + Pn[1]
515                if not self.mass_spectrum.is_centroid:
516                    mz_profile_calc = (Pn[0] * mz_exp_profile_tocal) + Pn[1]
517
518            elif order == 2:
519                mz_domain = (Pn[0] * (mz_exp_peaks_tocal)) + (
520                    Pn[1] * np.power((mz_exp_peaks_tocal), 2) + Pn[2]
521                )
522
523                if not self.mass_spectrum.is_centroid:
524                    mz_profile_calc = (Pn[0] * (mz_exp_profile_tocal)) + (
525                        Pn[1] * np.power((mz_exp_profile_tocal), 2) + Pn[2]
526                    )
527
528            if self.mzsegment:
529                # Recombine the mass domains
530                mz_domain = np.concatenate([mz_domain, mz_exp_peaks_unchanged])
531                mz_domain.sort()
532                if not self.mass_spectrum.is_centroid:
533                    mz_profile_calc = np.concatenate(
534                        [mz_profile_calc, mz_exp_profile_unchanged]
535                    )
536                    mz_profile_calc.sort()
537                # Sort them
538                if (
539                    mz_exp_peaks[0] > mz_exp_peaks[1]
540                ):  # If originally descending mass order
541                    mz_domain = mz_domain[::-1]
542                    if not self.mass_spectrum.is_centroid:
543                        mz_profile_calc = mz_profile_calc[::-1]
544
545            self.mass_spectrum.mz_cal = mz_domain
546            if not self.mass_spectrum.is_centroid:
547                self.mass_spectrum.mz_cal_profile = mz_profile_calc
548
549            self.mass_spectrum.calibration_order = order
550            self.mass_spectrum.calibration_RMS = float(res["fun"])
551            self.mass_spectrum.calibration_points = int(len(cal_refs_mz))
552            self.mass_spectrum.calibration_ref_mzs = cal_refs_mz
553            self.mass_spectrum.calibration_meas_mzs = cal_peaks_mz
554
555            self.mass_spectrum.calibration_segment = self.mzsegment
556
557            if diagnostic:
558                return self.mass_spectrum, res
559            return self.mass_spectrum
560        else:
561            warnings.warn("Too few calibration points - aborting.")
562            return self.mass_spectrum

Main recalibration function which uses a robust linear regression

This function performs the recalibration of the mass spectrum object. It iteratively applies

Parameters
  • cal_peaks_mz (list of float): masses of measured peaks to use in mass calibration.
  • cal_refs_mz (list of float): reference mz values of found calibration points.
  • order (int, optional): order of the recalibration function. 1 = linear, 2 = quadratic. The default is 1.
Returns
  • mass_spectrum (CoreMS mass spectrum object): Calibrated mass spectrum object
Notes

This function is adapted, in part, from the SPIKE project [1,2] and is based on the robust linear regression method.

References
  1. Chiron L., Coutouly M-A., Starck J-P., Rolando C., Delsuc M-A. SPIKE a Processing Software dedicated to Fourier Spectroscopies https://arxiv.org/abs/1608.06777 (2016)
  2. SPIKE - https://github.com/spike-project/spike
def run(self):
564    def run(self):
565        """Run the calibration routine
566
567        This function runs the calibration routine.
568
569        """
570        calib_snr_threshold = self.mass_spectrum.settings.calib_sn_threshold
571        max_calib_ppm_error = self.mass_spectrum.settings.max_calib_ppm_error
572        min_calib_ppm_error = self.mass_spectrum.settings.min_calib_ppm_error
573        calib_pol_order = self.mass_spectrum.settings.calib_pol_order
574        calibration_ref_match_method = (
575            self.mass_spectrum.settings.calibration_ref_match_method
576        )
577        calibration_ref_match_tolerance = (
578            self.mass_spectrum.settings.calibration_ref_match_tolerance
579        )
580        calibration_ref_match_std_raw_error_limit = (
581            self.mass_spectrum.settings.calibration_ref_match_std_raw_error_limit
582        )
583
584        # load reference mass list
585        df_ref = self.load_ref_mass_list()
586
587        # find calibration points
588        cal_peaks_mz, cal_refs_mz = self.find_calibration_points(
589            df_ref,
590            calib_ppm_error_threshold=(min_calib_ppm_error, max_calib_ppm_error),
591            calib_snr_threshold=calib_snr_threshold,
592            calibration_ref_match_method=calibration_ref_match_method,
593            calibration_ref_match_tolerance=calibration_ref_match_tolerance,
594            calibration_ref_match_std_raw_error_limit=calibration_ref_match_std_raw_error_limit,
595        )
596        if len(cal_peaks_mz) == 2:
597            self.mass_spectrum.settings.calib_pol_order = 1
598            calib_pol_order = 1
599            if self.mass_spectrum.parameters.mass_spectrum.verbose_processing:
600                print("Only 2 calibration points found, forcing a linear recalibration")
601        elif len(cal_peaks_mz) < 2:
602            warnings.warn("Too few calibration points found, function will fail")
603        self.recalibrate_mass_spectrum(cal_peaks_mz, cal_refs_mz, order=calib_pol_order)

Run the calibration routine

This function runs the calibration routine.