-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathChromFetcher2.py
107 lines (88 loc) · 4.07 KB
/
ChromFetcher2.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
from pathlib import Path
import struct
from xml.etree.ElementTree import parse
class Injection:
def __init__(self, path):
self.path = Path(path)
self.acqmethod = self._get_acq_method()
self.chromatograms = self._get_chromatograms()
self._set_data()
def _set_data(self):
for drug, transitions in self.acqmethod.items():
for trsn in transitions:
filt = self._filter(float(trsn['parent']),
float(trsn['fragment']),
float(trsn['fragmentor']),
float(trsn['collision energy']))
xy = [[i['time'], i['tic']] for i in filt]
trsn['points'] = xy
def _get_chromatograms(self) -> list[dict]:
scanbytes = self._open_injection()
magic_number = 186
offset = struct.unpack('I', scanbytes[88:92])[0]
n = len(scanbytes)
data = []
if (n - offset) % magic_number == 0:
nrows = (n - offset) // magic_number
for i in range(nrows):
start = offset + magic_number * i
data.append(self._entry(scanbytes[start:start + magic_number]))
return data
def _filter(self, parent, fragment, fragmentor, ce) -> list[dict]:
filt = [i for i in self.chromatograms if parent - .01 <= float(i['parent']) <= parent + .01]
filt = [i for i in filt if fragment - .01 <= float(i['fragment']) <= fragment + .01]
filt = [i for i in filt if float(fragmentor) == float(i['fragmentor'])]
filt = [i for i in filt if ce == i['collision energy']]
return filt
def _entry(self, blist: bytes) -> dict:
d = {'time': struct.unpack('d', blist[12:20])[0],
'tic': struct.unpack('d', blist[26:34])[0],
'fragment': struct.unpack('d', blist[34:42])[0],
'fragmentor': struct.unpack('f', blist[72:76])[0],
'collision energy': struct.unpack('f', blist[76:80])[0],
'parent': struct.unpack('d', blist[80:88])[0]}
return d
def _open_injection(self) -> bytes:
with open(self.path / 'AcqData\\MSScan.bin', 'rb') as file:
scanbytes = file.read()
return scanbytes
def _get_acq_method(self):
vals = {}
path = None
# Find XML Method file
method_files = self.path / 'AcqData'
for method_file in method_files.iterdir():
xpath = self.path / method_file / '192_1.xml'
if xpath.is_file():
path = xpath
# Parse xml data to dictionary 'vals'
if path:
x = parse(path)
root = x.getroot()
for child in root[10][0][5]:
for el in child:
drug, parent, fragment, fragmentor, ce = None, None, None, None, None
if el.tag == 'scanElements':
for grandchild in el[0]:
if grandchild.tag == 'compoundName':
drug = grandchild.text
elif grandchild.tag == 'ms1LowMz':
parent = grandchild.text
elif grandchild.tag == 'ms2LowMz':
fragment = grandchild.text
elif grandchild.tag == 'fragmentor':
fragmentor = grandchild.text
elif grandchild.tag == 'collisionEnergy':
ce = grandchild.text
val = [{'parent': parent,
'fragment': fragment,
'fragmentor': fragmentor,
'collision energy': ce,
'points': []}]
if drug in vals.keys():
vals[drug] = vals[drug] + val
else:
vals[drug] = val
return vals
sb = Injection('C:\\PinPoint\\MassHunter\\Data\\PAR\\Test Injections\\-\\Butalbital.d')
a=1