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

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): 
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            )

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.core.frame.DataFrame: 
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            clean_data = []
272            with self.file_location.open() as maglabfile:
273                for i in maglabfile.readlines()[8:-1]:
274                    clean_data.append(i.split())
275            dataframe = DataFrame(clean_data, columns=names)
276
277        elif self.data_type == "dataframe":
278            dataframe = read_pickle(data)
279
280        elif self.data_type == "excel":
281            dataframe = read_excel(data)
282
283        elif self.data_type == "xml":
284            dataframe = self.read_xml_peaks(data)
285
286        else:
287            raise TypeError("Data type %s is not supported" % self.data_type)
288
289        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): 
291    def load_settings(self, mass_spec_obj, output_parameters):
292        """
293        #TODO loading output parameters from json file is not functional
294        Load settings from a JSON file and apply them to the given mass_spec_obj.
295
296        Parameters
297        ----------
298        mass_spec_obj : MassSpec
299            The mass spectrum object to apply the settings to.
300
301        """
302        import warnings
303
304        settings_file_path = self.file_location.with_suffix(".json")
305
306        if settings_file_path.exists():
307            self._parameters = load_and_set_parameters_class(
308                "DataInput", self._parameters, parameters_path=settings_file_path
309            )
310
311            load_and_set_parameters_ms(
312                mass_spec_obj, parameters_path=settings_file_path
313            )
314
315        else:
316            warnings.warn(
317                "auto settings loading is enabled but could not locate the file:  %s. Please load the settings manually"
318                % settings_file_path
319            )
320
321        # TODO this will load the setting from SettingCoreMS.json
322        # coreMSHFD5 overrides this function to import the attrs stored in the h5 file
323        # loaded_settings = {}
324        # loaded_settings['MoleculaSearch'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MoleculaSearchSetting')
325        # loaded_settings['MassSpecPeak'] = self.get_scan_group_attr_data(scan_index,  time_index, 'MassSpecPeakSetting')
326
327        # loaded_settings['MassSpectrum'] = self.get_scan_group_attr_data(scan_index, time_index, 'MassSpectrumSetting')
328        # 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: 
330    def get_output_parameters(self, polarity: int, scan_index: int = 0) -> dict:
331        """
332        Get the output parameters for the mass spectrum.
333
334        Parameters
335        ----------
336        polarity : int
337            The polarity of the mass spectrum +1 or -1.
338        scan_index : int, optional
339            The index of the scan. Default is 0.
340
341        Returns
342        -------
343        dict
344            A dictionary containing the output parameters.
345
346        """
347
348        output_parameters = default_parameters(self.file_location)
349
350        if self.isCentroid:
351            output_parameters["label"] = Labels.corems_centroid
352        else:
353            output_parameters["label"] = Labels.bruker_profile
354
355        output_parameters["analyzer"] = self.analyzer
356
357        output_parameters["instrument_label"] = self.instrument_label
358
359        output_parameters["sample_name"] = self.sample_name
360
361        output_parameters["Aterm"] = None
362
363        output_parameters["Bterm"] = None
364
365        output_parameters["Cterm"] = None
366
367        output_parameters["polarity"] = polarity
368
369        # scan_number and rt will be need to lc ms====
370
371        output_parameters["mobility_scan"] = 0
372
373        output_parameters["mobility_rt"] = 0
374
375        output_parameters["scan_number"] = scan_index
376
377        output_parameters["rt"] = 0
378
379        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): 
381    def clean_data_frame(self, dataframe):
382        """
383        Clean the input dataframe by removing columns that are not expected.
384
385        Parameters
386        ----------
387        pandas.DataFrame
388            The input dataframe to be cleaned.
389
390        """
391
392        for column_name in dataframe.columns:
393            expected_column_name = self.parameters.header_translate.get(column_name)
394            if expected_column_name not in self._expected_columns:
395                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]): 
397    def check_columns(self, header_labels: list[str]):
398        """
399        Check if the given header labels match the expected columns.
400
401        Parameters
402        ----------
403        header_labels : list
404            The header labels to be checked.
405
406        Raises
407        ------
408        Exception
409            If any expected column is not found in the header labels.
410        """
411        found_label = set()
412
413        for label in header_labels:
414            if label not in self._expected_columns:
415                user_column_name = self.parameters.header_translate.get(label)
416                if user_column_name in self._expected_columns:
417                    found_label.add(user_column_name)
418            else:
419                found_label.add(label)
420
421        not_found = self._expected_columns - found_label
422
423        if len(not_found) > 0:
424            raise Exception(
425                "Please make sure to include the columns %s" % ", ".join(not_found)
426            )

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.core.frame.DataFrame: 
428    def read_xml_peaks(self, data: str) -> DataFrame:
429        """
430        Read peaks from a Bruker .xml file and return a pandas DataFrame.
431
432        Parameters
433        ----------
434        data : str
435            The path to the .xml file.
436
437        Returns
438        -------
439        pandas.DataFrame
440            A DataFrame containing the peak data with columns: 'm/z', 'I', 'Resolving Power', 'Area', 'S/N', 'fwhm'.
441        """
442        from numpy import nan
443
444        with open(data, "r") as file:
445            content = file.readlines()
446            content = "".join(content)
447            bs_content = BeautifulSoup(content, features="xml")
448        peaks_xml = bs_content.find_all("pk")
449
450        # initialise lists of the peak variables
451        areas = []
452        fwhms = []
453        intensities = []
454        mzs = []
455        res = []
456        sn = []
457        # iterate through the peaks appending to each list
458        for peak in peaks_xml:
459            areas.append(
460                float(peak.get("a", nan))
461            )  # Use a default value if key 'a' is missing
462            fwhms.append(
463                float(peak.get("fwhm", nan))
464            )  # Use a default value if key 'fwhm' is missing
465            intensities.append(
466                float(peak.get("i", nan))
467            )  # Use a default value if key 'i' is missing
468            mzs.append(
469                float(peak.get("mz", nan))
470            )  # Use a default value if key 'mz' is missing
471            res.append(
472                float(peak.get("res", nan))
473            )  # Use a default value if key 'res' is missing
474            sn.append(
475                float(peak.get("sn", nan))
476            )  # Use a default value if key 'sn' is missing
477
478        # Compile pandas dataframe of these values
479        names = ["m/z", "I", "Resolving Power", "Area", "S/N", "fwhm"]
480        df = DataFrame(columns=names, dtype=float)
481        df["m/z"] = mzs
482        df["I"] = intensities
483        df["Resolving Power"] = res
484        df["Area"] = areas
485        df["S/N"] = sn
486        df["fwhm"] = fwhms
487        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): 
489    def get_xml_polarity(self):
490        """
491        Get the polarity from an XML peaklist.
492
493        Returns
494        -------
495        int
496            The polarity of the XML peaklist. Returns -1 for negative polarity, +1 for positive polarity.
497
498        Raises
499        ------
500        Exception
501            If the data type is not XML peaklist in Bruker format or if the polarity is unhandled.
502        """
503
504        # Check its an actual xml
505        if not self.data_type or not self.delimiter:
506            self.set_data_type()
507
508        if isinstance(self.file_location, S3Path):
509            # data = self.file_location.open('rb').read()
510            data = BytesIO(self.file_location.open("rb").read())
511
512        else:
513            data = self.file_location
514
515        if self.data_type != "xml":
516            raise Exception("This function is only for XML peaklists (Bruker format)")
517
518        with open(data, "r") as file:
519            content = file.readlines()
520            content = "".join(content)
521            bs_content = BeautifulSoup(content, features="xml")
522        polarity = bs_content.find_all("ms_spectrum")[0]["polarity"]
523        if polarity == "-":
524            return -1
525        elif polarity == "+":
526            return +1
527        else:
528            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.