corems.transient.input.brukerSolarix

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Jun 12, 2019"
  3from copy import deepcopy
  4from datetime import datetime
  5from pathlib import Path
  6from xml.dom import minidom
  7
  8from numpy import dtype, float32, float64, frombuffer, fromfile, fromstring, genfromtxt
  9from s3path import S3Path
 10
 11from corems.encapsulation.factory.parameters import default_parameters
 12from corems.transient.factory.TransientClasses import Transient
 13
 14
 15class ReadBrukerSolarix(object):
 16    """A class used to Read a single Transient from Bruker's FT-MS acquisition station (fid, or ser)
 17
 18    Parameters
 19    ----------
 20    d_directory_location : str
 21        the full path of the .d folder
 22
 23    Attributes
 24    --------
 25    d_directory_location : str
 26        the full path of the .d folder
 27    file_location : str
 28        the full path of the .d folder
 29    parameter_filename_location : str
 30        the full path of the apexAcquisition.method file
 31    transient_data_path : str
 32        the full path of the fid or ser file
 33    scan_attr : str
 34        the full path of the scan.xml file
 35
 36
 37    Methods
 38    -------
 39    * get_transient().
 40        Read the data and settings returning a Transient class
 41    * get_scan_attr().
 42        Read the scan retention times, TIC values and scan indices.
 43    * locate_file(folder, type_file_name).
 44        Find the full path of a specific file within the acquisition .d folder or subfolders
 45    * parse_parameters(parameters_filename).
 46        Open the given file and retrieve all parameters from apexAcquisition.method
 47    * fix_freq_limits(d_parameters).
 48        Read and set the correct frequency limits for the spectrum
 49    * get_excite_sweep_range(filename).
 50        Determine excitation sweep range from ExciteSweep file
 51
 52    """
 53
 54    def __enter__(self):
 55        return self.get_transient()
 56
 57    def __exit__(self, exc_type, exc_val, exc_tb):
 58        return False
 59
 60    def __init__(self, d_directory_location):
 61        if isinstance(d_directory_location, str):
 62            d_directory_location = Path(d_directory_location)
 63
 64        if not d_directory_location.exists():
 65            raise FileNotFoundError("File does not exist: " + str(d_directory_location))
 66
 67        self.d_directory_location = d_directory_location
 68
 69        self.file_location = d_directory_location
 70
 71        try:
 72            self.parameter_filename_location = self.locate_file(
 73                d_directory_location, "apexAcquisition.method"
 74            )
 75            self.transient_data_path = d_directory_location / "fid"
 76
 77            if not self.transient_data_path.exists():
 78                self.transient_data_path = d_directory_location / "ser"
 79
 80                if not self.transient_data_path.exists():
 81                    raise FileNotFoundError("Could not locate transient data")
 82
 83                else:
 84                    # get scan attributes
 85                    self.scan_attr = d_directory_location / "scan.xml"
 86
 87        except:
 88            raise FileExistsError(
 89                "%s does not seem to be a valid Solarix Mass Spectrum"
 90                % (d_directory_location)
 91            )
 92
 93    def get_scan_attr(self):
 94        """Function to get the scan retention times, TIC values and scan indices.
 95
 96        Gets information from scan.xml file in the bruker .d folder.
 97        Note this file is only present in some .d format - e.g. for imaging mode data, it is not present.
 98
 99        Returns
100        -------
101        dict_scan_rt_tic : dict
102            a dictionary with scan number as key and rt and tic as values
103        """
104
105        from bs4 import BeautifulSoup
106
107        try:
108            soup = BeautifulSoup(self.scan_attr.open(), "xml")
109        except:
110            raise FileNotFoundError(
111                "Dataset does not appear to contain a 'scan.xml' file or it is misformated"
112            )
113
114        list_rt = [float(rt.text) for rt in soup.find_all("minutes")]
115        list_tic = [float(tic.text) for tic in soup.find_all("tic")]
116        list_scan = [int(scan.text) for scan in soup.find_all("count")]
117
118        dict_scan_rt_tic = dict(zip(list_scan, zip(list_rt, list_tic)))
119
120        return dict_scan_rt_tic
121
122    def get_transient(self, scan_number=1):
123        """Function to get the transient data and parameters from a Bruker Solarix .d folder.
124
125        Parameters
126        ----------
127        scan_number : int
128            the scan number to be read. Default is 1.
129
130        Returns
131        -------
132        Transient
133            a transient object
134        """
135
136        file_d_params = self.parse_parameters(self.parameter_filename_location)
137
138        self.fix_freq_limits(file_d_params)
139
140        from sys import platform
141
142        if platform == "win32":
143            # Windows...
144            dt = dtype("l")
145        else:
146            dt = dtype("i")
147
148        # get rt, scan, and tic from scan.xml file, otherwise  using 0 defaults values
149
150        output_parameters = deepcopy(default_parameters(self.d_directory_location))
151
152        if self.transient_data_path.name == "ser":
153            if self.scan_attr.exists():
154                dict_scan_rt_tic = self.get_scan_attr()
155
156                output_parameters["scan_number"] = scan_number
157
158                output_parameters["rt"] = dict_scan_rt_tic.get(scan_number)[0]
159
160                output_parameters["tic"] = dict_scan_rt_tic.get(scan_number)[1]
161
162        output_parameters["analyzer"] = "ICR"
163
164        output_parameters["label"] = "Bruker_Frequency"
165
166        output_parameters["Aterm"] = float(file_d_params.get("ML1"))
167
168        output_parameters["Bterm"] = float(file_d_params.get("ML2"))
169
170        output_parameters["Cterm"] = float(file_d_params.get("ML3"))
171
172        output_parameters["exc_high_freq"] = float(file_d_params.get("EXC_Freq_High"))
173
174        output_parameters["exc_low_freq"] = float(file_d_params.get("EXC_Freq_Low"))
175        try:
176            output_parameters["qpd_enabled"] = float(file_d_params.get("QPD_Enabled"))
177        except TypeError:  # for older datasets which dont have this variable
178            output_parameters["qpd_enabled"] = 0
179
180        output_parameters["mw_low"] = float(file_d_params.get("MW_low"))
181
182        output_parameters["mw_high"] = float(file_d_params.get("MW_high"))
183
184        output_parameters["bandwidth"] = float(file_d_params.get("SW_h"))
185
186        output_parameters["number_data_points"] = int(file_d_params.get("TD"))
187
188        output_parameters["polarity"] = str(file_d_params.get("Polarity"))
189
190        output_parameters["acquisition_time"] = file_d_params.get("acquisition_time")
191
192        data_points = int(file_d_params.get("TD"))
193
194        scan = output_parameters["scan_number"]
195        from io import BytesIO
196
197        if self.transient_data_path.name == "ser":
198            if isinstance(self.transient_data_path, S3Path):
199                databin = BytesIO(self.transient_data_path.open("rb").read())
200
201            else:
202                databin = self.transient_data_path.open("rb")
203
204            databin.seek((scan - 1) * 4 * data_points)
205            # read scan data and parse to 32int struct
206            data = frombuffer(databin.read(4 * data_points), dtype=dt)
207
208        else:
209            if isinstance(self.transient_data_path, S3Path):
210                data = frombuffer(self.transient_data_path.open("rb").read(), dtype=dt)
211            else:
212                data = fromfile(self.transient_data_path, dtype=dt)
213
214        return Transient(data, output_parameters)
215
216    #    for key, values in default_parameters.items():
217    #        print(key, values)
218    def fix_freq_limits(self, d_parameters):
219        """Function to read and set the correct frequency limits for the spectrum
220
221        Notes
222        --------
223        This is using the excitation limits from the apexAcquisition.method file,
224        which may not match the intended detection limits in edge cases.
225        In default acquisitions, excitation and detection are the same.
226        But, they may not be in some cases with selective excitation, custom excite waveforms, or in 2DMS applications.
227
228        Parameters
229        ----------
230        d_parameters : dict
231            a dictionary with the parameters from the apexAcquisition.method file
232        """
233
234        highfreq = float(d_parameters.get("EXC_Freq_High"))
235
236        lowfreq = float(d_parameters.get("EXC_Freq_Low"))
237
238        # CR for compatibility with Apex format as there is no EXciteSweep file
239        if not highfreq and lowfreq:
240            excitation_sweep_filelocation = self.locate_file(
241                self.d_directory_location, "ExciteSweep"
242            )
243            lowfreq, highfreq = self.get_excite_sweep_range(
244                excitation_sweep_filelocation
245            )
246            d_parameters["EXC_Freq_High"] = highfreq
247            d_parameters["EXC_Freq_Low"] = lowfreq
248
249    @staticmethod
250    def get_excite_sweep_range(filename):
251        """Function to determine excitation sweep range from ExciteSweep file
252
253        This looks at the first and last rows of the ExciteSweep file to determine the excitation frequency range.
254        Note that this assumes the excitation sweep was linear and the first and last rows are the lowest and highest frequencies.
255        This is presumably always true, but again may be incorrect for edge cases with custom excitation waveforms.
256
257        Parameters
258        ----------
259        filename : str
260            the full path to the ExciteSweep file
261
262        """
263        ExciteSweep_lines = genfromtxt(filename, comments="*", delimiter="\n")
264        # CR ready if we need the full array
265        highfreq = fromstring(ExciteSweep_lines[0])
266        lowfreq = fromstring(ExciteSweep_lines[-1])
267
268        return lowfreq[0], highfreq[0]
269
270    @staticmethod
271    def locate_file(folder, type_file_name="apexAcquisition.method"):
272        """Function to locate a file in a folder
273
274        Find the full path of a specific file within the acquisition .d folder or subfolders
275
276        Parameters
277        ----------
278        folder : str
279            the full path to the folder
280        type_file_name : str
281            the name of the file to be located
282            Expected options: ExciteSweep or apexAcquisition.method
283
284        Returns
285        -------
286        str
287            the full path to the file
288
289        Notes
290        -----
291        adapted from code from SPIKE library, https://github.com/spike-project/spike
292
293        """
294
295        from pathlib import Path
296
297        # directory_location = folder.glob( '**/*apexAcquisition.method')
298        directory_location = folder.glob("**/*" + type_file_name)
299        result = list(directory_location)
300        if len(result) > 1:
301            raise Exception(
302                "You have more than 1 %s file in the %s folder, using the first one"
303                % (type_file_name, folder)
304            )
305
306        elif len(result) == 0:
307            raise Exception(
308                "You don't have any %s file in the  %s folder, please double check the path"
309                % (type_file_name, folder)
310            )
311
312        return result[0]
313
314    @staticmethod
315    def parse_parameters(parameters_filename):
316        """Function to parse the parameters from apexAcquisition.method file
317
318        Open the given file and retrieve all parameters from apexAcquisition.method
319            None is written when no value for value is found
320
321            structure : <param name = "AMS_ActiveExclusion"><value>0</value></param>
322
323        Parameters
324        ----------
325        parameters_filename : str
326            the full path to the apexAcquisition.method file
327
328        Returns
329        -------
330        dict
331            a dictionary with the parameters and values
332
333        Notes
334        -----
335        Adapted from code from SPIKE library, https://github.com/spike-project/spike.
336        Code may not handle all possible parameters, but should be sufficient for most common use cases
337        """
338
339        # TODO: change to beautiful soup xml parsing
340
341        xmldoc = minidom.parse(parameters_filename.open())
342
343        x = xmldoc.documentElement
344        parameter_dict = {}
345        children = x.childNodes
346        for child in children:
347            # print( child.node)
348            if child.nodeName == "methodmetadata":
349                sections = child.childNodes
350                for section in sections:
351                    for element in section.childNodes:
352                        if element.nodeName == "date":
353                            # if element.nodeName == "primarykey":
354
355                            date_time_str = element.childNodes[0].nodeValue
356                            # parameter_dict["acquisition_time"] = pd.to_datetime(date_time_str, infer_datetime_format=True).to_pydatetime()
357                            parameter_dict["acquisition_time"] = datetime.strptime(
358                                date_time_str, "%b_%d_%Y %H:%M:%S.%f"
359                            )
360
361            if child.nodeName == "reportinfo":
362                sections = child.childNodes
363                for section in sections:
364                    if section.nodeName == "section":
365                        if section.getAttribute("title") == "Main":
366                            for element in section.childNodes:
367                                if element.nodeName == "section":
368                                    if element.getAttribute("title") == "Polarity":
369                                        if (
370                                            str(
371                                                element.childNodes[1].getAttribute(
372                                                    "value"
373                                                )
374                                            )
375                                            == "Negative"
376                                        ):
377                                            parameter_dict["Polarity"] = -1
378                                        else:
379                                            parameter_dict["Polarity"] = 1
380
381            if child.nodeName == "paramlist":
382                params = child.childNodes
383                for param in params:
384                    # print( param.nodeName)
385                    if param.nodeName == "param":
386                        paramenter_label = str(param.getAttribute("name"))
387                        for element in param.childNodes:
388                            if element.nodeName == "value":
389                                try:
390                                    parameter_value = str(element.firstChild.toxml())
391                                    # print v
392                                except:
393                                    parameter_value = None
394
395                            parameter_dict[paramenter_label] = parameter_value
396
397        return parameter_dict
398
399    def parse_sqlite(self, sqlite_filename="chromatography-data.sqlite"):
400        """ """
401        import sqlite3
402
403        def read_sqlite_file(file_path, table_name):
404            """
405            Read data from a SQLite database file and return it as a list of tuples
406
407            Parameters
408            ----------
409            file_path : str
410                the full path to the SQLite database file
411            table_name : str
412                the name of the table to be read
413
414            Returns
415            -------
416            list
417                a list of tuples with the data from the table
418            """
419            # Connect to the SQLite database file
420            conn = sqlite3.connect(file_path)
421            cursor = conn.cursor()
422
423            # Execute a query to select data from a table (replace 'table_name' with your table's name)
424            query = f"SELECT * FROM {table_name}"
425            cursor.execute(query)
426
427            # Fetch all rows from the result set
428            rows = cursor.fetchall()
429            stream = []
430            # Print or process the fetched rows
431            for row in rows:
432                stream.append(row)
433                # print(row)  # Print each row, you can also process it differently
434
435            # Close the cursor and the connection
436            cursor.close()
437            conn.close()
438            return stream
439
440        def parse_binary(binary, type):
441            """
442            Parse binary data from the sqlite data streams
443            """
444            if type == "double":
445                data = frombuffer(binary, dtype=float64)
446            elif type == "float":
447                data = frombuffer(binary, dtype=float32)
448            return data
449
450        sqlite_filelocation = self.locate_file(
451            self.d_directory_location, sqlite_filename
452        )
453        table_name = "TraceSources"
454        trace_sources = read_sqlite_file(sqlite_filelocation, table_name)
455        table_name = "TraceChunks"
456        trace_chunks = read_sqlite_file(sqlite_filelocation, table_name)
457        times = []
458        values = []
459        trace_type = {}
460
461        for index, source in enumerate(trace_sources):
462            trace_id = source[0]
463            trace_type[source[1]] = {"times": [], "values": []}
464            for index, chunk in enumerate(trace_chunks):
465                id = chunk[0]
466                times = parse_binary(chunk[1], "double")
467                values = parse_binary(chunk[2], "float")
468                for time, value in zip(times, values):
469                    if source[0] == id:
470                        trace_type[source[1]]["times"].append(time)
471                        trace_type[source[1]]["values"].append(value)
472
473        return trace_type