diff --git a/flowio/flowdata.py b/flowio/flowdata.py index b3188e9..a458c6e 100644 --- a/flowio/flowdata.py +++ b/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) diff --git a/flowio/tests/flowdata_tests.py b/flowio/tests/flowdata_tests.py index 6e585b0..677f4ad 100644 --- a/flowio/tests/flowdata_tests.py +++ b/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: