Merge branch 'JcGKitten-master'

Author: whitews

flowio/flowdata.py

@@ -29,8 +29,10 @@ class FlowData(object):
 
     :param filename_or_handle: a path string or a file handle for an FCS file
     :param ignore_offset_error: option to ignore data offset error (see above note), default is False
+    :param only_text: option to only read the "text" segment of the FCS file without loading event data,
+        default is False
     """
-    def __init__(self, filename_or_handle, ignore_offset_error=False):
+    def __init__(self, filename_or_handle, ignore_offset_error=False, only_text=False):
         if isinstance(filename_or_handle, basestring):
             self._fh = open(str(filename_or_handle), 'rb')
         else:
@@ -86,12 +88,15 @@ def __init__(self, filename_or_handle, ignore_offset_error=False):
         if d_stop > self.file_size:
             raise EOFError("FCS header indicates data section greater than file size")
 
-        self.events = self.__parse_data(
-            self.cur_offset,
-            d_start,
-            d_stop,
-            self.text
-        )
+        if only_text:
+            self.events = None
+        else:
+            self.events = self.__parse_data(
+                self.cur_offset,
+                d_start,
+                d_stop,
+                self.text
+            )
 
         self.channels = self._parse_channels()
 
@@ -377,6 +382,12 @@ def write_fcs(self, filename, extra=None, extra_non_standard=None):
         :param extra_non_standard: an optional dictionary for adding extra non-standard keywords/values
         :return: None
         """
+        if self.events is None:
+            raise AttributeError(
+                "FlowData instance does not contain event data. This might"
+                "occur if the FCS file was read with the only_text=True option."
+            )
+
         pnn_labels = [''] * len(self.channels)
         pns_labels = [''] * len(self.channels)
 

flowio/tests/flowdata_tests.py

@@ -27,6 +27,12 @@ def test_event_count(self):
     def test_get_text(self):
         self.assertEqual(self.flow_data.text['cyt'], 'FACScan')
 
+    def test_load_only_text(self):
+        flow_data = FlowData('examples/fcs_files/3FITC_4PE_004.fcs', only_text=True)
+
+        self.assertIsNone(flow_data.events)
+        self.assertRaises(AttributeError, flow_data.write_fcs, 'delete_this_file.fcs')
+
     @staticmethod
     def test_load_fcs_from_memory():
         with open('examples/fcs_files/3FITC_4PE_004.fcs', 'rb') as f: