Add FAIMS splitting script

Author: mobiusklein

examples/faims_split.py

@@ -0,0 +1,57 @@
+import click
+import ms_deisotope
+from ms_deisotope import output
+
+from ms_deisotope.output import MzMLSerializer
+from ms_deisotope.data_source import query
+
+from ms_deisotope.tools.utils import register_debug_hook
+
+
+@click.command('faims-split')
+@click.argument("source_file", type=click.Path(exists=True, readable=True))
+@click.argument("output_prefix")
+def main(source_file, output_prefix):
+    reader = ms_deisotope.MSFileLoader(source_file)
+
+    sinks = {}
+    n = len(reader)
+    i = 0
+    last = 0
+    interval = 1000
+    demultiplexer = query.FAIMSDemultiplexingIterator(reader, 30)
+
+    for heads in demultiplexer:
+        empty_channels = 0
+        for channel, value in heads.items():
+            if value is not None:
+                if channel not in sinks:
+                    click.echo("Opening new channel for CV %r" % (channel, ))
+                    writer = MzMLSerializer(
+                        open(output_prefix + "_%r.mzML" % (channel, ), 'wb'),
+                        len(reader), deconvoluted=False)
+                    writer.copy_metadata_from(reader)
+                    method = writer.build_processing_method(1, False, False, False, ["ion mobility seperation"])
+                    writer.add_data_processing(method)
+                    sinks[channel] = writer
+                writer.save_scan(value)
+                i += 1
+            else:
+                empty_channels += 1
+        if empty_channels == len(heads):
+            click.echo("All channels empty, finishing")
+            break
+        if i - last >= interval:
+            click.echo("Processed %d spectra (%0.3f%%)" % (i, i * 100.0 / n))
+            last = i
+
+    click.echo("Closing buffers.")
+    for sink in sinks.values():
+        click.echo("Closing %r" % (sink, ))
+        sink.close()
+
+if __name__ == "__main__":
+    main.main()
+
+
+

ms_deisotope/data_source/query.py

@@ -598,6 +598,10 @@ def _check(self, scan):
                 if scan.drift_time == self.compensation_voltage:
                     return scan
 
+    @property
+    def key(self):
+        return self.compensation_voltage
+
 
 class QueueIterator(ScanIteratorProxyBase):
     def __init__(self, scan_source, data=None):
@@ -623,7 +627,7 @@ def __init__(self, scan_source, buffer_size=10, *args, **kwargs):
             scan_source, *args, **kwargs)
         self.channels = dict()
         self.buffer_size = buffer_size
-        self.feed()
+        self.has_more = self.feed()
 
     @property
     def iteration_mode(self):
@@ -667,13 +671,15 @@ def feed(self):
         return True
 
     def __next__(self):
+        if not self.has_more:
+            raise StopIteration()
         meta = {}
         for channel_name, (scan_filter, queue) in self.channels.items():
             if queue.has_value():
                 meta[channel_name] = next(scan_filter)
             else:
                 meta[channel_name] = None
-        self.feed()
+        self.has_more = self.feed()
         return meta
 
 

ms_deisotope/data_source/scan/base.py

@@ -1625,10 +1625,12 @@ def annotate_precursor(self, *args, **kwargs):
         --------
         :func:`ms_deisotope.plot.annotate_scan_single`
         '''
-        pinfo = self.scan.precursor_information
-        if pinfo is None:
-            return
-        precursor = pinfo.precursor
+        precursor = kwargs.get("precursor")
+        if precursor is None:
+            pinfo = self.scan.precursor_information
+            if pinfo is None:
+                return
+            precursor = pinfo.precursor
         return self._plot_api.annotate_scan_single(precursor, self.scan, *args, **kwargs)
 
     def label_peaks(self, *args, **kwargs):

ms_deisotope/output/mzml.py

@@ -986,7 +986,7 @@ def extract_scan_event_parameters(self, scan):
                 })
             traits = scan_event.traits.items()
             for name, value in traits:
-                param = {"name": name, "value": value, 'unit_name': getattr(value, 'unit_info', None)}
+                param = {"name": str(name), "value": value, 'unit_name': getattr(value, 'unit_info', None)}
                 scan_parameters.append(param)
             scan_window_list = list(scan_event)
         return scan_parameters, scan_window_list