update portals for .mzml and .raw files

Author: RJMW

dimspy/portals/mzml_portal.py

@@ -113,10 +113,21 @@ def tics(self):
         # print self.run()[2]
         for scan in self.run():
             if scan["id"] == "TIC":
-                tics = zip(*scan.peaks)[1]
-                return tics
+                return zip(*scan.peaks)[1]
         return
 
+    def injection_times(self):
+        injection_times = {}
+        for scan in self.run():
+            injection_times[scan['id']] = None
+            for element in scan.xmlTree:
+                if "MS:1000927" == element.get('accession'):
+                    injection_times[scan['id']] = float(element.get("value"))
+                    break
+            if scan['id'] not in injection_times:
+                injection_times[scan['id']] = None
+        return injection_times
+
     def scan_dependents(self):
         l = []
         for scan in self.run():

dimspy/portals/thermo_raw_portal.py

@@ -24,66 +24,53 @@
 
 
 def mz_range_from_header(h):
-    """
-    Extract a list of headers / .
-    :rtype: list
-    """
     return [float(m) for m in re.findall(r'([\w\.-]+)-([\w\.-]+)', h)[0]]
 
 
 class ThermoRaw:
-    """
-    Extract a list of headers / .
-    :rtype: list
-    """
+
     def __init__(self, filename):
         self.run = RawFileReader.RawFileReaderAdapter.FileFactory(filename)
         self.run.SelectInstrument(Business.Device.MS, 1)
+        self.filename = filename
 
     def headers(self):
-        """
-        Extract a particular scan from a *.raw file and return a PeakList objects
-        :rtype: dict
-        """
+
         sids = collections.OrderedDict()
         for scan_id in range(self.run.RunHeaderEx.FirstSpectrum, self.run.RunHeaderEx.LastSpectrum + 1):
             sids.setdefault(str(self.run.GetFilterForScanNumber(scan_id).Filter), []).append(scan_id)
         return sids
 
     def scan_ids(self):
-        """
-        Extract a particular scan from a *.raw file and return a PeakList objects
-        :rtype: dict
-        """
+
         sids = collections.OrderedDict()
         for scan_id in range(self.run.RunHeaderEx.FirstSpectrum, self.run.RunHeaderEx.LastSpectrum + 1):
             sids[scan_id] = str(self.run.GetFilterForScanNumber(scan_id).Filter)
         return sids
 
     def peaklist(self, scan_id, function_noise="noise_packets"):
-        """
-        Extract a particular scan from a *.raw file and return a PeakList objects
 
-        :param scan_ids:
-        :rtype: list
-        """
         if function_noise not in ["noise_packets", "mean", "median", "mad"]:
             raise ValueError("select a function that is available [noise_packets, mean, median, mad]")
 
         scan = self.run.GetCentroidStream(scan_id, False)
-
-        mz_ibn = zip(scan.Masses, scan.Intensities, scan.Baselines, scan.Noises)  # SignalToNoise not available
-        mz_ibn.sort()
-        mzs, ints, baseline, noise = zip(*mz_ibn)
+        if scan.Masses is not None:
+            mz_ibn = zip(scan.Masses, scan.Intensities, scan.Baselines, scan.Noises)  # SignalToNoise not available
+            mz_ibn.sort()
+            mzs, ints, baseline, noise = zip(*mz_ibn)
+        else:
+            mzs, ints, baseline, noise = [], [], [], []
 
         if function_noise == "noise_packets":
             snr = [p.SignalToNoise for p in scan.GetCentroids()]
-        elif function_noise == "median":
+        elif function_noise == "median" and len(ints) > 0:
             snr = ints / np.median(ints)
-        elif function_noise == "mean":
+        elif function_noise == "mean" and len(ints) > 0:
             snr = ints / np.mean(ints)
-        elif function_noise == "mad":
+        elif function_noise == "mad" and len(ints) > 0:
             snr = ints / np.median(np.abs(np.subtract(ints, np.median(ints))))
+        else:
+            snr = []
 
         scan_stats = self.run.GetScanStatsForScanNumber(scan_id)
 
@@ -119,39 +106,43 @@ def peaklist(self, scan_id, function_noise="noise_packets"):
                       tic=tic,
                       function_noise=function_noise)
 
-        pl.add_attribute('snr', snr)
-        pl.add_attribute('noise', noise)
-        pl.add_attribute('baseline', baseline)
+        if len(pl.mz) > 0:
+            pl.add_attribute('snr', snr)
+            pl.add_attribute('noise', noise)
+            pl.add_attribute('baseline', baseline)
+
         return pl
 
     def peaklists(self, scan_ids, function_noise="noise_packets"):
-        """
-        Extract the scans from a *.raw file and return a list of PeakList objects
-
-        :param scan_ids:
-        :rtype: list
-
-        """
         if function_noise not in ["noise_packets", "mean", "median", "mad"]:
             raise ValueError("select a function that is available [noise_packets, mean, median, mad]")
 
         return [self.peaklist(scan_id, function_noise=function_noise) for scan_id in scan_ids]
 
     def tics(self):
-        # somehow i can not access the scans directly when run() uses an open archive object
-        # print self.run()[2]
-        tics = []
+        tics = collections.OrderedDict()
         for scan_id in range(self.run.RunHeaderEx.FirstSpectrum, self.run.RunHeaderEx.LastSpectrum + 1):
             scan_stats = self.run.GetScanStatsForScanNumber(scan_id)
-            tics.append(scan_stats.TIC)
+            tics[scan_id].append(scan_stats.TIC)
         return tics
 
+    def injection_times(self):
+        injection_times = collections.OrderedDict()
+        for scan_id in range(self.run.RunHeaderEx.FirstSpectrum, self.run.RunHeaderEx.LastSpectrum + 1):
+            extra_values = list(self.run.GetTrailerExtraInformation(scan_id).Values)
+            extra_labels = list(self.run.GetTrailerExtraInformation(scan_id).Labels)
+            for i, label in enumerate(extra_labels):
+                if "Ion Injection Time (ms):" == label:
+                    injection_times[scan_id] = float(extra_values[i])
+            if scan_id not in injection_times:
+                injection_times[scan_id] = None
+        return injection_times
+
     def scan_dependents(self):
         l = []
         for scan_id in range(self.run.RunHeaderEx.FirstSpectrum, self.run.RunHeaderEx.LastSpectrum + 1):
             gsd = self.run.GetScanDependents(scan_id, 5)
             if gsd is not None:
                 for i, d in enumerate(gsd.ScanDependentDetailArray):
-                    print scan_id, self.run.GetFilterForScanNumber(scan_id).Filter, d.ScanIndex, d.FilterString
                     l.append([scan_id, d.ScanIndex])
         return l