corems.mass_spectrum.input.baseClass 

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Nov 11, 2019"
  3
  4from copy import deepcopy
  5from io import BytesIO
  6from pathlib import Path
  7
  8import chardet
  9from bs4 import BeautifulSoup
 10from pandas import read_csv, read_excel, read_pickle
 11from pandas.core.frame import DataFrame
 12from s3path import S3Path
 13
 14from corems.encapsulation.constant import Labels
 15from corems.encapsulation.factory.parameters import default_parameters
 16from corems.encapsulation.factory.processingSetting import DataInputSetting
 17from corems.encapsulation.input.parameter_from_json import (
 18    load_and_set_parameters_class,
 19    load_and_set_parameters_ms,
 20    load_and_set_toml_parameters_class,
 21)
 22
 23
 24class MassListBaseClass:
 25    """The MassListBaseClass object reads mass list data types and returns the mass spectrum obj
 26
 27    Parameters
 28    ----------
 29    file_location : Path or S3Path
 30        Full data path.
 31    isCentroid : bool, optional
 32        Determines the mass spectrum data structure. If set to True, it assumes centroid mode. If set to False, it assumes profile mode and attempts to peak pick. Default is True.
 33    analyzer : str, optional
 34        The analyzer used for the mass spectrum. Default is 'Unknown'.
 35    instrument_label : str, optional
 36        The label of the instrument used for the mass spectrum. Default is 'Unknown'.
 37    sample_name : str, optional
 38        The name of the sample. Default is None.
 39    header_lines : int, optional
 40        The number of lines to skip in the file, including the column labels line. Default is 0.
 41    isThermoProfile : bool, optional
 42        Determines the number of expected columns in the file. If set to True, only m/z and intensity columns are expected. Signal-to-noise ratio (S/N) and resolving power (RP) will be calculated based on the data. Default is False.
 43    headerless : bool, optional
 44        If True, assumes that there are no headers present in the file (e.g., a .xy file from Bruker) and assumes two columns: m/z and intensity. Default is False.
 45
 46    Attributes
 47    ----------
 48    parameters : DataInputSetting
 49        The data input settings for the mass spectrum.
 50    data_type : str
 51        The type of data in the file.
 52    delimiter : str
 53        The delimiter used to read text-based files.
 54
 55    Methods
 56    -------
 57    * set_parameter_from_toml(parameters_path). Sets the data input settings from a TOML file.
 58    * set_parameter_from_json(parameters_path). Sets the data input settings from a JSON file.
 59    * get_dataframe(). Reads the file and returns the data as a pandas DataFrame.
 60    * load_settings(mass_spec_obj, output_parameters). Loads the settings for the mass spectrum.
 61    * get_output_parameters(polarity, scan_index=0). Returns the output parameters for the mass spectrum.
 62    * clean_data_frame(dataframe). Cleans the data frame by removing columns that are not in the expected columns set.
 63
 64    """
 65
 66    def __init__(
 67        self,
 68        file_location: Path | S3Path,
 69        isCentroid: bool = True,
 70        analyzer: str = "Unknown",
 71        instrument_label: str = "Unknown",
 72        sample_name: str = None,
 73        header_lines: int = 0,
 74        isThermoProfile: bool = False,
 75        headerless: bool = False,
 76    ):
 77        self.file_location = (
 78            Path(file_location) if isinstance(file_location, str) else file_location
 79        )
 80
 81        if not self.file_location.exists():
 82            raise FileExistsError("File does not exist: %s" % file_location)
 83
 84        # (newline="\n")
 85
 86        self.header_lines = header_lines
 87
 88        if isThermoProfile:
 89            self._expected_columns = {Labels.mz, Labels.abundance}
 90
 91        else:
 92            self._expected_columns = {
 93                Labels.mz,
 94                Labels.abundance,
 95                Labels.s2n,
 96                Labels.rp,
 97            }
 98
 99        self._delimiter = None
100
101        self.isCentroid = isCentroid
102
103        self.isThermoProfile = isThermoProfile
104
105        self.headerless = headerless
106
107        self._data_type = None
108
109        self.analyzer = analyzer
110
111        self.instrument_label = instrument_label
112
113        self.sample_name = sample_name
114
115        self._parameters = deepcopy(DataInputSetting())
116
117    @property
118    def parameters(self):
119        return self._parameters
120
121    @parameters.setter
122    def parameters(self, instance_DataInputSetting):
123        self._parameters = instance_DataInputSetting
124
125    def set_parameter_from_toml(self, parameters_path):
126        self._parameters = load_and_set_toml_parameters_class(
127            "DataInput", self.parameters, parameters_path=parameters_path
128        )
129
130    def set_parameter_from_json(self, parameters_path):
131        self._parameters = load_and_set_parameters_class(
132            "DataInput", self.parameters, parameters_path=parameters_path
133        )
134
135    @property
136    def data_type(self):
137        return self._data_type
138
139    @data_type.setter
140    def data_type(self, data_type):
141        self._data_type = data_type
142
143    @property
144    def delimiter(self):
145        return self._delimiter
146
147    @delimiter.setter
148    def delimiter(self, delimiter):
149        self._delimiter = delimiter
150
151    def encoding_detector(self, file_location) -> str:
152        """
153        Detects the encoding of a file.
154
155        Parameters
156        --------
157        file_location : str
158            The location of the file to be analyzed.
159
160        Returns
161        --------
162        str
163            The detected encoding of the file.
164        """
165
166        with file_location.open("rb") as rawdata:
167            result = chardet.detect(rawdata.read(10000))
168        return result["encoding"]
169
170    def set_data_type(self):
171        """
172        Set the data type and delimiter based on the file extension.
173
174        Raises
175        ------
176        TypeError
177            If the data type could not be automatically recognized.
178        """
179        if self.file_location.suffix == ".csv":
180            self.data_type = "txt"
181            self.delimiter = ","
182        elif self.file_location.suffix == ".txt":
183            self.data_type = "txt"
184            self.delimiter = "\t"
185        elif self.file_location.suffix == ".tsv":
186            self.data_type = "txt"
187            self.delimiter = "\t"
188        elif self.file_location.suffix == ".xlsx":
189            self.data_type = "excel"
190        elif self.file_location.suffix == ".ascii":
191            self.data_type = "txt"
192            self.delimiter = "  "
193        elif self.file_location.suffix == ".pkl":
194            self.data_type = "dataframe"
195        elif self.file_location.suffix == ".pks":
196            self.data_type = "pks"
197            self.delimiter = "          "
198            self.header_lines = 9
199        elif self.file_location.suffix == ".xml":
200            self.data_type = "xml"
201            # self.delimiter = None
202            # self.header_lines = None
203        elif self.file_location.suffix == ".xy":
204            self.data_type = "txt"
205            self.delimiter = " "
206            self.header_lines = None
207        else:
208            raise TypeError(
209                "Data type could not be automatically recognized for %s; please set data type and delimiter manually."
210                % self.file_location.name
211            )
212
213    def get_dataframe(self) -> DataFrame:
214        """
215        Get the data as a pandas DataFrame.
216
217        Returns
218        -------
219        pandas.DataFrame
220            The data as a pandas DataFrame.
221
222        Raises
223        ------
224        TypeError
225            If the data type is not supported.
226        """
227
228        if not self.data_type or not self.delimiter:
229            self.set_data_type()
230
231        if isinstance(self.file_location, S3Path):
232            data = BytesIO(self.file_location.open("rb").read())
233        else:
234            data = self.file_location
235
236        if self.data_type == "txt":
237            if self.headerless:
238                dataframe = read_csv(
239                    data,
240                    skiprows=self.header_lines,
241                    delimiter=self.delimiter,
242                    header=None,
243                    names=["m/z", "I"],
244                    encoding=self.encoding_detector(self.file_location),
245                    engine="python",
246                )
247            else:
248                dataframe = read_csv(
249                    data,
250                    skiprows=self.header_lines,
251                    delimiter=self.delimiter,
252                    encoding=self.encoding_detector(self.file_location),
253                    engine="python",
254                )
255
256        elif self.data_type == "pks":
257            # Predator .pks columns are positional: peak location (m/z), relative
258            # peak height (normalized 0-100), absolute abundance, resolving power,
259            # frequency, S/N. Use the absolute abundance as the intensity -- the
260            # relative peak height is per-spectrum normalized and not comparable
261            # across spectra, so it is named so header_translate drops it.
262            names = [
263                "m/z",
264                "Relative Abundance",
265                "Abundance",
266                "Resolving Power",
267                "Frequency",
268                "S/N"
269                ]
270
271        
272            clean_data = []
273            with self.file_location.open() as maglabfile:
274                for i in maglabfile.readlines()[8:-1]:
275                    clean_data.append(i.split())
276            dataframe = DataFrame(clean_data, columns=names)
277
278        elif self.data_type == "dataframe":
279            dataframe = read_pickle(data)
280
281        elif self.data_type == "excel":
282            dataframe = read_excel(data)
283
284        elif self.data_type == "xml":
285            dataframe = self.read_xml_peaks(data)
286
287        else:
288            raise TypeError("Data type %s is not supported" % self.data_type)
289
290        return dataframe
291
292    def load_settings(self, mass_spec_obj, output_parameters):
293        """
294        #TODO loading output parameters from json file is not functional
295        Load settings from a JSON file and apply them to the given mass_spec_obj.
296
297        Parameters
298        ----------
299        mass_spec_obj : MassSpec
300            The mass spectrum object to apply the settings to.
301
302        """
303        import json
304        import warnings
305
306        settings_file_path = self.file_location.with_suffix(".json")
307
308        if settings_file_path.exists():
309            self._parameters = load_and_set_parameters_class(
310                "DataInput", self._parameters, parameters_path=settings_file_path
311            )
312
313            load_and_set_parameters_ms(
314                mass_spec_obj, parameters_path=settings_file_path
315            )
316
317        else:
318            warnings.warn(
319                "auto settings loading is enabled but could not locate the file:  %s. Please load the settings manually"
320                % settings_file_path
321            )
322
323        # TODO this will load the setting from SettingCoreMS.json
324        # coreMSHFD5 overrides this function to import the attrs stored in the h5 file
325        # loaded_settings = {}
326        # loaded_settings['MoleculaSearch'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MoleculaSearchSetting')
327        # loaded_settings['MassSpecPeak'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MassSpecPeakSetting')
328
329        # loaded_settings['MassSpectrum'] = self.get_scan_group_attr_data(scan_index, time_index, 'MassSpectrumSetting')
330        # loaded_settings['Transient'] = self.get_scan_group_attr_data(scan_index, time_index, 'TransientSetting')
331
332    def get_output_parameters(self, polarity: int, scan_index: int = 0) -> dict:
333        """
334        Get the output parameters for the mass spectrum.
335
336        Parameters
337        ----------
338        polarity : int
339            The polarity of the mass spectrum +1 or -1.
340        scan_index : int, optional
341            The index of the scan. Default is 0.
342
343        Returns
344        -------
345        dict
346            A dictionary containing the output parameters.
347
348        """
349        from copy import deepcopy
350
351        output_parameters = default_parameters(self.file_location)
352
353        if self.isCentroid:
354            output_parameters["label"] = Labels.corems_centroid
355        else:
356            output_parameters["label"] = Labels.bruker_profile
357
358        output_parameters["analyzer"] = self.analyzer
359
360        output_parameters["instrument_label"] = self.instrument_label
361
362        output_parameters["sample_name"] = self.sample_name
363
364        output_parameters["Aterm"] = None
365
366        output_parameters["Bterm"] = None
367
368        output_parameters["Cterm"] = None
369
370        output_parameters["polarity"] = polarity
371
372        # scan_number and rt will be need to lc ms====
373
374        output_parameters["mobility_scan"] = 0
375
376        output_parameters["mobility_rt"] = 0
377
378        output_parameters["scan_number"] = scan_index
379
380        output_parameters["rt"] = 0
381
382        return output_parameters
383
384    def clean_data_frame(self, dataframe):
385        """
386        Clean the input dataframe by removing columns that are not expected.
387
388        Parameters
389        ----------
390        pandas.DataFrame
391            The input dataframe to be cleaned.
392
393        """
394
395        for column_name in dataframe.columns:
396            expected_column_name = self.parameters.header_translate.get(column_name)
397            if expected_column_name not in self._expected_columns:
398                del dataframe[column_name]
399
400    def check_columns(self, header_labels: list[str]):
401        """
402        Check if the given header labels match the expected columns.
403
404        Parameters
405        ----------
406        header_labels : list
407            The header labels to be checked.
408
409        Raises
410        ------
411        Exception
412            If any expected column is not found in the header labels.
413        """
414        found_label = set()
415
416        for label in header_labels:
417            if not label in self._expected_columns:
418                user_column_name = self.parameters.header_translate.get(label)
419                if user_column_name in self._expected_columns:
420                    found_label.add(user_column_name)
421            else:
422                found_label.add(label)
423
424        not_found = self._expected_columns - found_label
425
426        if len(not_found) > 0:
427            raise Exception(
428                "Please make sure to include the columns %s" % ", ".join(not_found)
429            )
430
431    def read_xml_peaks(self, data: str) -> DataFrame:
432        """
433        Read peaks from a Bruker .xml file and return a pandas DataFrame.
434
435        Parameters
436        ----------
437        data : str
438            The path to the .xml file.
439
440        Returns
441        -------
442        pandas.DataFrame
443            A DataFrame containing the peak data with columns: 'm/z', 'I', 'Resolving Power', 'Area', 'S/N', 'fwhm'.
444        """
445        from numpy import nan
446
447        with open(data, "r") as file:
448            content = file.readlines()
449            content = "".join(content)
450            bs_content = BeautifulSoup(content, features="xml")
451        peaks_xml = bs_content.find_all("pk")
452
453        # initialise lists of the peak variables
454        areas = []
455        fwhms = []
456        intensities = []
457        mzs = []
458        res = []
459        sn = []
460        # iterate through the peaks appending to each list
461        for peak in peaks_xml:
462            areas.append(
463                float(peak.get("a", nan))
464            )  # Use a default value if key 'a' is missing
465            fwhms.append(
466                float(peak.get("fwhm", nan))
467            )  # Use a default value if key 'fwhm' is missing
468            intensities.append(
469                float(peak.get("i", nan))
470            )  # Use a default value if key 'i' is missing
471            mzs.append(
472                float(peak.get("mz", nan))
473            )  # Use a default value if key 'mz' is missing
474            res.append(
475                float(peak.get("res", nan))
476            )  # Use a default value if key 'res' is missing
477            sn.append(
478                float(peak.get("sn", nan))
479            )  # Use a default value if key 'sn' is missing
480
481        # Compile pandas dataframe of these values
482        names = ["m/z", "I", "Resolving Power", "Area", "S/N", "fwhm"]
483        df = DataFrame(columns=names, dtype=float)
484        df["m/z"] = mzs
485        df["I"] = intensities
486        df["Resolving Power"] = res
487        df["Area"] = areas
488        df["S/N"] = sn
489        df["fwhm"] = fwhms
490        return df
491
492    def get_xml_polarity(self):
493        """
494        Get the polarity from an XML peaklist.
495
496        Returns
497        -------
498        int
499            The polarity of the XML peaklist. Returns -1 for negative polarity, +1 for positive polarity.
500
501        Raises
502        ------
503        Exception
504            If the data type is not XML peaklist in Bruker format or if the polarity is unhandled.
505        """
506
507        # Check its an actual xml
508        if not self.data_type or not self.delimiter:
509            self.set_data_type()
510
511        if isinstance(self.file_location, S3Path):
512            # data = self.file_location.open('rb').read()
513            data = BytesIO(self.file_location.open("rb").read())
514
515        else:
516            data = self.file_location
517
518        if self.data_type != "xml":
519            raise Exception("This function is only for XML peaklists (Bruker format)")
520
521        with open(data, "r") as file:
522            content = file.readlines()
523            content = "".join(content)
524            bs_content = BeautifulSoup(content, features="xml")
525        polarity = bs_content.find_all("ms_spectrum")[0]["polarity"]
526        if polarity == "-":
527            return -1
528        elif polarity == "+":
529            return +1
530        else:
531            raise Exception("Polarity %s unhandled" % polarity)
class MassListBaseClass: 
 25class MassListBaseClass:
 26    """The MassListBaseClass object reads mass list data types and returns the mass spectrum obj
 27
 28    Parameters
 29    ----------
 30    file_location : Path or S3Path
 31        Full data path.
 32    isCentroid : bool, optional
 33        Determines the mass spectrum data structure. If set to True, it assumes centroid mode. If set to False, it assumes profile mode and attempts to peak pick. Default is True.
 34    analyzer : str, optional
 35        The analyzer used for the mass spectrum. Default is 'Unknown'.
 36    instrument_label : str, optional
 37        The label of the instrument used for the mass spectrum. Default is 'Unknown'.
 38    sample_name : str, optional
 39        The name of the sample. Default is None.
 40    header_lines : int, optional
 41        The number of lines to skip in the file, including the column labels line. Default is 0.
 42    isThermoProfile : bool, optional
 43        Determines the number of expected columns in the file. If set to True, only m/z and intensity columns are expected. Signal-to-noise ratio (S/N) and resolving power (RP) will be calculated based on the data. Default is False.
 44    headerless : bool, optional
 45        If True, assumes that there are no headers present in the file (e.g., a .xy file from Bruker) and assumes two columns: m/z and intensity. Default is False.
 46
 47    Attributes
 48    ----------
 49    parameters : DataInputSetting
 50        The data input settings for the mass spectrum.
 51    data_type : str
 52        The type of data in the file.
 53    delimiter : str
 54        The delimiter used to read text-based files.
 55
 56    Methods
 57    -------
 58    * set_parameter_from_toml(parameters_path). Sets the data input settings from a TOML file.
 59    * set_parameter_from_json(parameters_path). Sets the data input settings from a JSON file.
 60    * get_dataframe(). Reads the file and returns the data as a pandas DataFrame.
 61    * load_settings(mass_spec_obj, output_parameters). Loads the settings for the mass spectrum.
 62    * get_output_parameters(polarity, scan_index=0). Returns the output parameters for the mass spectrum.
 63    * clean_data_frame(dataframe). Cleans the data frame by removing columns that are not in the expected columns set.
 64
 65    """
 66
 67    def __init__(
 68        self,
 69        file_location: Path | S3Path,
 70        isCentroid: bool = True,
 71        analyzer: str = "Unknown",
 72        instrument_label: str = "Unknown",
 73        sample_name: str = None,
 74        header_lines: int = 0,
 75        isThermoProfile: bool = False,
 76        headerless: bool = False,
 77    ):
 78        self.file_location = (
 79            Path(file_location) if isinstance(file_location, str) else file_location
 80        )
 81
 82        if not self.file_location.exists():
 83            raise FileExistsError("File does not exist: %s" % file_location)
 84
 85        # (newline="\n")
 86
 87        self.header_lines = header_lines
 88
 89        if isThermoProfile:
 90            self._expected_columns = {Labels.mz, Labels.abundance}
 91
 92        else:
 93            self._expected_columns = {
 94                Labels.mz,
 95                Labels.abundance,
 96                Labels.s2n,
 97                Labels.rp,
 98            }
 99
100        self._delimiter = None
101
102        self.isCentroid = isCentroid
103
104        self.isThermoProfile = isThermoProfile
105
106        self.headerless = headerless
107
108        self._data_type = None
109
110        self.analyzer = analyzer
111
112        self.instrument_label = instrument_label
113
114        self.sample_name = sample_name
115
116        self._parameters = deepcopy(DataInputSetting())
117
118    @property
119    def parameters(self):
120        return self._parameters
121
122    @parameters.setter
123    def parameters(self, instance_DataInputSetting):
124        self._parameters = instance_DataInputSetting
125
126    def set_parameter_from_toml(self, parameters_path):
127        self._parameters = load_and_set_toml_parameters_class(
128            "DataInput", self.parameters, parameters_path=parameters_path
129        )
130
131    def set_parameter_from_json(self, parameters_path):
132        self._parameters = load_and_set_parameters_class(
133            "DataInput", self.parameters, parameters_path=parameters_path
134        )
135
136    @property
137    def data_type(self):
138        return self._data_type
139
140    @data_type.setter
141    def data_type(self, data_type):
142        self._data_type = data_type
143
144    @property
145    def delimiter(self):
146        return self._delimiter
147
148    @delimiter.setter
149    def delimiter(self, delimiter):
150        self._delimiter = delimiter
151
152    def encoding_detector(self, file_location) -> str:
153        """
154        Detects the encoding of a file.
155
156        Parameters
157        --------
158        file_location : str
159            The location of the file to be analyzed.
160
161        Returns
162        --------
163        str
164            The detected encoding of the file.
165        """
166
167        with file_location.open("rb") as rawdata:
168            result = chardet.detect(rawdata.read(10000))
169        return result["encoding"]
170
171    def set_data_type(self):
172        """
173        Set the data type and delimiter based on the file extension.
174
175        Raises
176        ------
177        TypeError
178            If the data type could not be automatically recognized.
179        """
180        if self.file_location.suffix == ".csv":
181            self.data_type = "txt"
182            self.delimiter = ","
183        elif self.file_location.suffix == ".txt":
184            self.data_type = "txt"
185            self.delimiter = "\t"
186        elif self.file_location.suffix == ".tsv":
187            self.data_type = "txt"
188            self.delimiter = "\t"
189        elif self.file_location.suffix == ".xlsx":
190            self.data_type = "excel"
191        elif self.file_location.suffix == ".ascii":
192            self.data_type = "txt"
193            self.delimiter = "  "
194        elif self.file_location.suffix == ".pkl":
195            self.data_type = "dataframe"
196        elif self.file_location.suffix == ".pks":
197            self.data_type = "pks"
198            self.delimiter = "          "
199            self.header_lines = 9
200        elif self.file_location.suffix == ".xml":
201            self.data_type = "xml"
202            # self.delimiter = None
203            # self.header_lines = None
204        elif self.file_location.suffix == ".xy":
205            self.data_type = "txt"
206            self.delimiter = " "
207            self.header_lines = None
208        else:
209            raise TypeError(
210                "Data type could not be automatically recognized for %s; please set data type and delimiter manually."
211                % self.file_location.name
212            )
213
214    def get_dataframe(self) -> DataFrame:
215        """
216        Get the data as a pandas DataFrame.
217
218        Returns
219        -------
220        pandas.DataFrame
221            The data as a pandas DataFrame.
222
223        Raises
224        ------
225        TypeError
226            If the data type is not supported.
227        """
228
229        if not self.data_type or not self.delimiter:
230            self.set_data_type()
231
232        if isinstance(self.file_location, S3Path):
233            data = BytesIO(self.file_location.open("rb").read())
234        else:
235            data = self.file_location
236
237        if self.data_type == "txt":
238            if self.headerless:
239                dataframe = read_csv(
240                    data,
241                    skiprows=self.header_lines,
242                    delimiter=self.delimiter,
243                    header=None,
244                    names=["m/z", "I"],
245                    encoding=self.encoding_detector(self.file_location),
246                    engine="python",
247                )
248            else:
249                dataframe = read_csv(
250                    data,
251                    skiprows=self.header_lines,
252                    delimiter=self.delimiter,
253                    encoding=self.encoding_detector(self.file_location),
254                    engine="python",
255                )
256
257        elif self.data_type == "pks":
258            # Predator .pks columns are positional: peak location (m/z), relative
259            # peak height (normalized 0-100), absolute abundance, resolving power,
260            # frequency, S/N. Use the absolute abundance as the intensity -- the
261            # relative peak height is per-spectrum normalized and not comparable
262            # across spectra, so it is named so header_translate drops it.
263            names = [
264                "m/z",
265                "Relative Abundance",
266                "Abundance",
267                "Resolving Power",
268                "Frequency",
269                "S/N"
270                ]
271
272        
273            clean_data = []
274            with self.file_location.open() as maglabfile:
275                for i in maglabfile.readlines()[8:-1]:
276                    clean_data.append(i.split())
277            dataframe = DataFrame(clean_data, columns=names)
278
279        elif self.data_type == "dataframe":
280            dataframe = read_pickle(data)
281
282        elif self.data_type == "excel":
283            dataframe = read_excel(data)
284
285        elif self.data_type == "xml":
286            dataframe = self.read_xml_peaks(data)
287
288        else:
289            raise TypeError("Data type %s is not supported" % self.data_type)
290
291        return dataframe
292
293    def load_settings(self, mass_spec_obj, output_parameters):
294        """
295        #TODO loading output parameters from json file is not functional
296        Load settings from a JSON file and apply them to the given mass_spec_obj.
297
298        Parameters
299        ----------
300        mass_spec_obj : MassSpec
301            The mass spectrum object to apply the settings to.
302
303        """
304        import json
305        import warnings
306
307        settings_file_path = self.file_location.with_suffix(".json")
308
309        if settings_file_path.exists():
310            self._parameters = load_and_set_parameters_class(
311                "DataInput", self._parameters, parameters_path=settings_file_path
312            )
313
314            load_and_set_parameters_ms(
315                mass_spec_obj, parameters_path=settings_file_path
316            )
317
318        else:
319            warnings.warn(
320                "auto settings loading is enabled but could not locate the file:  %s. Please load the settings manually"
321                % settings_file_path
322            )
323
324        # TODO this will load the setting from SettingCoreMS.json
325        # coreMSHFD5 overrides this function to import the attrs stored in the h5 file
326        # loaded_settings = {}
327        # loaded_settings['MoleculaSearch'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MoleculaSearchSetting')
328        # loaded_settings['MassSpecPeak'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MassSpecPeakSetting')
329
330        # loaded_settings['MassSpectrum'] = self.get_scan_group_attr_data(scan_index, time_index, 'MassSpectrumSetting')
331        # loaded_settings['Transient'] = self.get_scan_group_attr_data(scan_index, time_index, 'TransientSetting')
332
333    def get_output_parameters(self, polarity: int, scan_index: int = 0) -> dict:
334        """
335        Get the output parameters for the mass spectrum.
336
337        Parameters
338        ----------
339        polarity : int
340            The polarity of the mass spectrum +1 or -1.
341        scan_index : int, optional
342            The index of the scan. Default is 0.
343
344        Returns
345        -------
346        dict
347            A dictionary containing the output parameters.
348
349        """
350        from copy import deepcopy
351
352        output_parameters = default_parameters(self.file_location)
353
354        if self.isCentroid:
355            output_parameters["label"] = Labels.corems_centroid
356        else:
357            output_parameters["label"] = Labels.bruker_profile
358
359        output_parameters["analyzer"] = self.analyzer
360
361        output_parameters["instrument_label"] = self.instrument_label
362
363        output_parameters["sample_name"] = self.sample_name
364
365        output_parameters["Aterm"] = None
366
367        output_parameters["Bterm"] = None
368
369        output_parameters["Cterm"] = None
370
371        output_parameters["polarity"] = polarity
372
373        # scan_number and rt will be need to lc ms====
374
375        output_parameters["mobility_scan"] = 0
376
377        output_parameters["mobility_rt"] = 0
378
379        output_parameters["scan_number"] = scan_index
380
381        output_parameters["rt"] = 0
382
383        return output_parameters
384
385    def clean_data_frame(self, dataframe):
386        """
387        Clean the input dataframe by removing columns that are not expected.
388
389        Parameters
390        ----------
391        pandas.DataFrame
392            The input dataframe to be cleaned.
393
394        """
395
396        for column_name in dataframe.columns:
397            expected_column_name = self.parameters.header_translate.get(column_name)
398            if expected_column_name not in self._expected_columns:
399                del dataframe[column_name]
400
401    def check_columns(self, header_labels: list[str]):
402        """
403        Check if the given header labels match the expected columns.
404
405        Parameters
406        ----------
407        header_labels : list
408            The header labels to be checked.
409
410        Raises
411        ------
412        Exception
413            If any expected column is not found in the header labels.
414        """
415        found_label = set()
416
417        for label in header_labels:
418            if not label in self._expected_columns:
419                user_column_name = self.parameters.header_translate.get(label)
420                if user_column_name in self._expected_columns:
421                    found_label.add(user_column_name)
422            else:
423                found_label.add(label)
424
425        not_found = self._expected_columns - found_label
426
427        if len(not_found) > 0:
428            raise Exception(
429                "Please make sure to include the columns %s" % ", ".join(not_found)
430            )
431
432    def read_xml_peaks(self, data: str) -> DataFrame:
433        """
434        Read peaks from a Bruker .xml file and return a pandas DataFrame.
435
436        Parameters
437        ----------
438        data : str
439            The path to the .xml file.
440
441        Returns
442        -------
443        pandas.DataFrame
444            A DataFrame containing the peak data with columns: 'm/z', 'I', 'Resolving Power', 'Area', 'S/N', 'fwhm'.
445        """
446        from numpy import nan
447
448        with open(data, "r") as file:
449            content = file.readlines()
450            content = "".join(content)
451            bs_content = BeautifulSoup(content, features="xml")
452        peaks_xml = bs_content.find_all("pk")
453
454        # initialise lists of the peak variables
455        areas = []
456        fwhms = []
457        intensities = []
458        mzs = []
459        res = []
460        sn = []
461        # iterate through the peaks appending to each list
462        for peak in peaks_xml:
463            areas.append(
464                float(peak.get("a", nan))
465            )  # Use a default value if key 'a' is missing
466            fwhms.append(
467                float(peak.get("fwhm", nan))
468            )  # Use a default value if key 'fwhm' is missing
469            intensities.append(
470                float(peak.get("i", nan))
471            )  # Use a default value if key 'i' is missing
472            mzs.append(
473                float(peak.get("mz", nan))
474            )  # Use a default value if key 'mz' is missing
475            res.append(
476                float(peak.get("res", nan))
477            )  # Use a default value if key 'res' is missing
478            sn.append(
479                float(peak.get("sn", nan))
480            )  # Use a default value if key 'sn' is missing
481
482        # Compile pandas dataframe of these values
483        names = ["m/z", "I", "Resolving Power", "Area", "S/N", "fwhm"]
484        df = DataFrame(columns=names, dtype=float)
485        df["m/z"] = mzs
486        df["I"] = intensities
487        df["Resolving Power"] = res
488        df["Area"] = areas
489        df["S/N"] = sn
490        df["fwhm"] = fwhms
491        return df
492
493    def get_xml_polarity(self):
494        """
495        Get the polarity from an XML peaklist.
496
497        Returns
498        -------
499        int
500            The polarity of the XML peaklist. Returns -1 for negative polarity, +1 for positive polarity.
501
502        Raises
503        ------
504        Exception
505            If the data type is not XML peaklist in Bruker format or if the polarity is unhandled.
506        """
507
508        # Check its an actual xml
509        if not self.data_type or not self.delimiter:
510            self.set_data_type()
511
512        if isinstance(self.file_location, S3Path):
513            # data = self.file_location.open('rb').read()
514            data = BytesIO(self.file_location.open("rb").read())
515
516        else:
517            data = self.file_location
518
519        if self.data_type != "xml":
520            raise Exception("This function is only for XML peaklists (Bruker format)")
521
522        with open(data, "r") as file:
523            content = file.readlines()
524            content = "".join(content)
525            bs_content = BeautifulSoup(content, features="xml")
526        polarity = bs_content.find_all("ms_spectrum")[0]["polarity"]
527        if polarity == "-":
528            return -1
529        elif polarity == "+":
530            return +1
531        else:
532            raise Exception("Polarity %s unhandled" % polarity)

The MassListBaseClass object reads mass list data types and returns the mass spectrum obj

Parameters
  • file_location (Path or S3Path): Full data path.
  • isCentroid (bool, optional): Determines the mass spectrum data structure. If set to True, it assumes centroid mode. If set to False, it assumes profile mode and attempts to peak pick. Default is True.
  • analyzer (str, optional): The analyzer used for the mass spectrum. Default is 'Unknown'.
  • instrument_label (str, optional): The label of the instrument used for the mass spectrum. Default is 'Unknown'.
  • sample_name (str, optional): The name of the sample. Default is None.
  • header_lines (int, optional): The number of lines to skip in the file, including the column labels line. Default is 0.
  • isThermoProfile (bool, optional): Determines the number of expected columns in the file. If set to True, only m/z and intensity columns are expected. Signal-to-noise ratio (S/N) and resolving power (RP) will be calculated based on the data. Default is False.
  • headerless (bool, optional): If True, assumes that there are no headers present in the file (e.g., a .xy file from Bruker) and assumes two columns: m/z and intensity. Default is False.
Attributes
  • parameters (DataInputSetting): The data input settings for the mass spectrum.
  • data_type (str): The type of data in the file.
  • delimiter (str): The delimiter used to read text-based files.
Methods
  • set_parameter_from_toml(parameters_path). Sets the data input settings from a TOML file.
  • set_parameter_from_json(parameters_path). Sets the data input settings from a JSON file.
  • get_dataframe(). Reads the file and returns the data as a pandas DataFrame.
  • load_settings(mass_spec_obj, output_parameters). Loads the settings for the mass spectrum.
  • get_output_parameters(polarity, scan_index=0). Returns the output parameters for the mass spectrum.
  • clean_data_frame(dataframe). Cleans the data frame by removing columns that are not in the expected columns set.
MassListBaseClass( file_location: pathlib._local.Path | s3path.current_version.S3Path, isCentroid: bool = True, analyzer: str = 'Unknown', instrument_label: str = 'Unknown', sample_name: str = None, header_lines: int = 0, isThermoProfile: bool = False, headerless: bool = False) 
 67    def __init__(
 68        self,
 69        file_location: Path | S3Path,
 70        isCentroid: bool = True,
 71        analyzer: str = "Unknown",
 72        instrument_label: str = "Unknown",
 73        sample_name: str = None,
 74        header_lines: int = 0,
 75        isThermoProfile: bool = False,
 76        headerless: bool = False,
 77    ):
 78        self.file_location = (
 79            Path(file_location) if isinstance(file_location, str) else file_location
 80        )
 81
 82        if not self.file_location.exists():
 83            raise FileExistsError("File does not exist: %s" % file_location)
 84
 85        # (newline="\n")
 86
 87        self.header_lines = header_lines
 88
 89        if isThermoProfile:
 90            self._expected_columns = {Labels.mz, Labels.abundance}
 91
 92        else:
 93            self._expected_columns = {
 94                Labels.mz,
 95                Labels.abundance,
 96                Labels.s2n,
 97                Labels.rp,
 98            }
 99
100        self._delimiter = None
101
102        self.isCentroid = isCentroid
103
104        self.isThermoProfile = isThermoProfile
105
106        self.headerless = headerless
107
108        self._data_type = None
109
110        self.analyzer = analyzer
111
112        self.instrument_label = instrument_label
113
114        self.sample_name = sample_name
115
116        self._parameters = deepcopy(DataInputSetting())
file_location
header_lines
isCentroid
isThermoProfile
headerless
analyzer
instrument_label
sample_name
parameters 
118    @property
119    def parameters(self):
120        return self._parameters
def set_parameter_from_toml(self, parameters_path): 
126    def set_parameter_from_toml(self, parameters_path):
127        self._parameters = load_and_set_toml_parameters_class(
128            "DataInput", self.parameters, parameters_path=parameters_path
129        )
def set_parameter_from_json(self, parameters_path): 
131    def set_parameter_from_json(self, parameters_path):
132        self._parameters = load_and_set_parameters_class(
133            "DataInput", self.parameters, parameters_path=parameters_path
134        )
data_type 
136    @property
137    def data_type(self):
138        return self._data_type
delimiter 
144    @property
145    def delimiter(self):
146        return self._delimiter
def encoding_detector(self, file_location) -> str: 
152    def encoding_detector(self, file_location) -> str:
153        """
154        Detects the encoding of a file.
155
156        Parameters
157        --------
158        file_location : str
159            The location of the file to be analyzed.
160
161        Returns
162        --------
163        str
164            The detected encoding of the file.
165        """
166
167        with file_location.open("rb") as rawdata:
168            result = chardet.detect(rawdata.read(10000))
169        return result["encoding"]

Detects the encoding of a file.

Parameters
  • file_location (str): The location of the file to be analyzed.
Returns
  • str: The detected encoding of the file.
def set_data_type(self): 
171    def set_data_type(self):
172        """
173        Set the data type and delimiter based on the file extension.
174
175        Raises
176        ------
177        TypeError
178            If the data type could not be automatically recognized.
179        """
180        if self.file_location.suffix == ".csv":
181            self.data_type = "txt"
182            self.delimiter = ","
183        elif self.file_location.suffix == ".txt":
184            self.data_type = "txt"
185            self.delimiter = "\t"
186        elif self.file_location.suffix == ".tsv":
187            self.data_type = "txt"
188            self.delimiter = "\t"
189        elif self.file_location.suffix == ".xlsx":
190            self.data_type = "excel"
191        elif self.file_location.suffix == ".ascii":
192            self.data_type = "txt"
193            self.delimiter = "  "
194        elif self.file_location.suffix == ".pkl":
195            self.data_type = "dataframe"
196        elif self.file_location.suffix == ".pks":
197            self.data_type = "pks"
198            self.delimiter = "          "
199            self.header_lines = 9
200        elif self.file_location.suffix == ".xml":
201            self.data_type = "xml"
202            # self.delimiter = None
203            # self.header_lines = None
204        elif self.file_location.suffix == ".xy":
205            self.data_type = "txt"
206            self.delimiter = " "
207            self.header_lines = None
208        else:
209            raise TypeError(
210                "Data type could not be automatically recognized for %s; please set data type and delimiter manually."
211                % self.file_location.name
212            )

Set the data type and delimiter based on the file extension.

Raises
  • TypeError: If the data type could not be automatically recognized.
def get_dataframe(self) -> pandas.DataFrame: 
214    def get_dataframe(self) -> DataFrame:
215        """
216        Get the data as a pandas DataFrame.
217
218        Returns
219        -------
220        pandas.DataFrame
221            The data as a pandas DataFrame.
222
223        Raises
224        ------
225        TypeError
226            If the data type is not supported.
227        """
228
229        if not self.data_type or not self.delimiter:
230            self.set_data_type()
231
232        if isinstance(self.file_location, S3Path):
233            data = BytesIO(self.file_location.open("rb").read())
234        else:
235            data = self.file_location
236
237        if self.data_type == "txt":
238            if self.headerless:
239                dataframe = read_csv(
240                    data,
241                    skiprows=self.header_lines,
242                    delimiter=self.delimiter,
243                    header=None,
244                    names=["m/z", "I"],
245                    encoding=self.encoding_detector(self.file_location),
246                    engine="python",
247                )
248            else:
249                dataframe = read_csv(
250                    data,
251                    skiprows=self.header_lines,
252                    delimiter=self.delimiter,
253                    encoding=self.encoding_detector(self.file_location),
254                    engine="python",
255                )
256
257        elif self.data_type == "pks":
258            # Predator .pks columns are positional: peak location (m/z), relative
259            # peak height (normalized 0-100), absolute abundance, resolving power,
260            # frequency, S/N. Use the absolute abundance as the intensity -- the
261            # relative peak height is per-spectrum normalized and not comparable
262            # across spectra, so it is named so header_translate drops it.
263            names = [
264                "m/z",
265                "Relative Abundance",
266                "Abundance",
267                "Resolving Power",
268                "Frequency",
269                "S/N"
270                ]
271
272        
273            clean_data = []
274            with self.file_location.open() as maglabfile:
275                for i in maglabfile.readlines()[8:-1]:
276                    clean_data.append(i.split())
277            dataframe = DataFrame(clean_data, columns=names)
278
279        elif self.data_type == "dataframe":
280            dataframe = read_pickle(data)
281
282        elif self.data_type == "excel":
283            dataframe = read_excel(data)
284
285        elif self.data_type == "xml":
286            dataframe = self.read_xml_peaks(data)
287
288        else:
289            raise TypeError("Data type %s is not supported" % self.data_type)
290
291        return dataframe

Get the data as a pandas DataFrame.

Returns
  • pandas.DataFrame: The data as a pandas DataFrame.
Raises
  • TypeError: If the data type is not supported.
def load_settings(self, mass_spec_obj, output_parameters): 
293    def load_settings(self, mass_spec_obj, output_parameters):
294        """
295        #TODO loading output parameters from json file is not functional
296        Load settings from a JSON file and apply them to the given mass_spec_obj.
297
298        Parameters
299        ----------
300        mass_spec_obj : MassSpec
301            The mass spectrum object to apply the settings to.
302
303        """
304        import json
305        import warnings
306
307        settings_file_path = self.file_location.with_suffix(".json")
308
309        if settings_file_path.exists():
310            self._parameters = load_and_set_parameters_class(
311                "DataInput", self._parameters, parameters_path=settings_file_path
312            )
313
314            load_and_set_parameters_ms(
315                mass_spec_obj, parameters_path=settings_file_path
316            )
317
318        else:
319            warnings.warn(
320                "auto settings loading is enabled but could not locate the file:  %s. Please load the settings manually"
321                % settings_file_path
322            )
323
324        # TODO this will load the setting from SettingCoreMS.json
325        # coreMSHFD5 overrides this function to import the attrs stored in the h5 file
326        # loaded_settings = {}
327        # loaded_settings['MoleculaSearch'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MoleculaSearchSetting')
328        # loaded_settings['MassSpecPeak'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MassSpecPeakSetting')
329
330        # loaded_settings['MassSpectrum'] = self.get_scan_group_attr_data(scan_index, time_index, 'MassSpectrumSetting')
331        # loaded_settings['Transient'] = self.get_scan_group_attr_data(scan_index, time_index, 'TransientSetting')

TODO loading output parameters from json file is not functional

Load settings from a JSON file and apply them to the given mass_spec_obj.

Parameters
  • mass_spec_obj (MassSpec): The mass spectrum object to apply the settings to.
def get_output_parameters(self, polarity: int, scan_index: int = 0) -> dict: 
333    def get_output_parameters(self, polarity: int, scan_index: int = 0) -> dict:
334        """
335        Get the output parameters for the mass spectrum.
336
337        Parameters
338        ----------
339        polarity : int
340            The polarity of the mass spectrum +1 or -1.
341        scan_index : int, optional
342            The index of the scan. Default is 0.
343
344        Returns
345        -------
346        dict
347            A dictionary containing the output parameters.
348
349        """
350        from copy import deepcopy
351
352        output_parameters = default_parameters(self.file_location)
353
354        if self.isCentroid:
355            output_parameters["label"] = Labels.corems_centroid
356        else:
357            output_parameters["label"] = Labels.bruker_profile
358
359        output_parameters["analyzer"] = self.analyzer
360
361        output_parameters["instrument_label"] = self.instrument_label
362
363        output_parameters["sample_name"] = self.sample_name
364
365        output_parameters["Aterm"] = None
366
367        output_parameters["Bterm"] = None
368
369        output_parameters["Cterm"] = None
370
371        output_parameters["polarity"] = polarity
372
373        # scan_number and rt will be need to lc ms====
374
375        output_parameters["mobility_scan"] = 0
376
377        output_parameters["mobility_rt"] = 0
378
379        output_parameters["scan_number"] = scan_index
380
381        output_parameters["rt"] = 0
382
383        return output_parameters

Get the output parameters for the mass spectrum.

Parameters
  • polarity (int): The polarity of the mass spectrum +1 or -1.
  • scan_index (int, optional): The index of the scan. Default is 0.
Returns
  • dict: A dictionary containing the output parameters.
def clean_data_frame(self, dataframe): 
385    def clean_data_frame(self, dataframe):
386        """
387        Clean the input dataframe by removing columns that are not expected.
388
389        Parameters
390        ----------
391        pandas.DataFrame
392            The input dataframe to be cleaned.
393
394        """
395
396        for column_name in dataframe.columns:
397            expected_column_name = self.parameters.header_translate.get(column_name)
398            if expected_column_name not in self._expected_columns:
399                del dataframe[column_name]

Clean the input dataframe by removing columns that are not expected.

Parameters
  • pandas.DataFrame: The input dataframe to be cleaned.
def check_columns(self, header_labels: list[str]): 
401    def check_columns(self, header_labels: list[str]):
402        """
403        Check if the given header labels match the expected columns.
404
405        Parameters
406        ----------
407        header_labels : list
408            The header labels to be checked.
409
410        Raises
411        ------
412        Exception
413            If any expected column is not found in the header labels.
414        """
415        found_label = set()
416
417        for label in header_labels:
418            if not label in self._expected_columns:
419                user_column_name = self.parameters.header_translate.get(label)
420                if user_column_name in self._expected_columns:
421                    found_label.add(user_column_name)
422            else:
423                found_label.add(label)
424
425        not_found = self._expected_columns - found_label
426
427        if len(not_found) > 0:
428            raise Exception(
429                "Please make sure to include the columns %s" % ", ".join(not_found)
430            )

Check if the given header labels match the expected columns.

Parameters
  • header_labels (list): The header labels to be checked.
Raises
  • Exception: If any expected column is not found in the header labels.
def read_xml_peaks(self, data: str) -> pandas.DataFrame: 
432    def read_xml_peaks(self, data: str) -> DataFrame:
433        """
434        Read peaks from a Bruker .xml file and return a pandas DataFrame.
435
436        Parameters
437        ----------
438        data : str
439            The path to the .xml file.
440
441        Returns
442        -------
443        pandas.DataFrame
444            A DataFrame containing the peak data with columns: 'm/z', 'I', 'Resolving Power', 'Area', 'S/N', 'fwhm'.
445        """
446        from numpy import nan
447
448        with open(data, "r") as file:
449            content = file.readlines()
450            content = "".join(content)
451            bs_content = BeautifulSoup(content, features="xml")
452        peaks_xml = bs_content.find_all("pk")
453
454        # initialise lists of the peak variables
455        areas = []
456        fwhms = []
457        intensities = []
458        mzs = []
459        res = []
460        sn = []
461        # iterate through the peaks appending to each list
462        for peak in peaks_xml:
463            areas.append(
464                float(peak.get("a", nan))
465            )  # Use a default value if key 'a' is missing
466            fwhms.append(
467                float(peak.get("fwhm", nan))
468            )  # Use a default value if key 'fwhm' is missing
469            intensities.append(
470                float(peak.get("i", nan))
471            )  # Use a default value if key 'i' is missing
472            mzs.append(
473                float(peak.get("mz", nan))
474            )  # Use a default value if key 'mz' is missing
475            res.append(
476                float(peak.get("res", nan))
477            )  # Use a default value if key 'res' is missing
478            sn.append(
479                float(peak.get("sn", nan))
480            )  # Use a default value if key 'sn' is missing
481
482        # Compile pandas dataframe of these values
483        names = ["m/z", "I", "Resolving Power", "Area", "S/N", "fwhm"]
484        df = DataFrame(columns=names, dtype=float)
485        df["m/z"] = mzs
486        df["I"] = intensities
487        df["Resolving Power"] = res
488        df["Area"] = areas
489        df["S/N"] = sn
490        df["fwhm"] = fwhms
491        return df

Read peaks from a Bruker .xml file and return a pandas DataFrame.

Parameters
  • data (str): The path to the .xml file.
Returns
  • pandas.DataFrame: A DataFrame containing the peak data with columns: 'm/z', 'I', 'Resolving Power', 'Area', 'S/N', 'fwhm'.
def get_xml_polarity(self): 
493    def get_xml_polarity(self):
494        """
495        Get the polarity from an XML peaklist.
496
497        Returns
498        -------
499        int
500            The polarity of the XML peaklist. Returns -1 for negative polarity, +1 for positive polarity.
501
502        Raises
503        ------
504        Exception
505            If the data type is not XML peaklist in Bruker format or if the polarity is unhandled.
506        """
507
508        # Check its an actual xml
509        if not self.data_type or not self.delimiter:
510            self.set_data_type()
511
512        if isinstance(self.file_location, S3Path):
513            # data = self.file_location.open('rb').read()
514            data = BytesIO(self.file_location.open("rb").read())
515
516        else:
517            data = self.file_location
518
519        if self.data_type != "xml":
520            raise Exception("This function is only for XML peaklists (Bruker format)")
521
522        with open(data, "r") as file:
523            content = file.readlines()
524            content = "".join(content)
525            bs_content = BeautifulSoup(content, features="xml")
526        polarity = bs_content.find_all("ms_spectrum")[0]["polarity"]
527        if polarity == "-":
528            return -1
529        elif polarity == "+":
530            return +1
531        else:
532            raise Exception("Polarity %s unhandled" % polarity)

Get the polarity from an XML peaklist.

Returns
  • int: The polarity of the XML peaklist. Returns -1 for negative polarity, +1 for positive polarity.
Raises
  • Exception: If the data type is not XML peaklist in Bruker format or if the polarity is unhandled.