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