more asserts -> errors

Author: zm711

neo/core/baseneo.py

@@ -35,6 +35,24 @@ class MergeError(Exception):
     pass
 
 class NeoReadWriteError(IOError):
+    """
+    This is the main neo-specific error that has to deal with
+    ephys data that does not follow the requirements for the Neo
+    object hierarchy.
+
+    It should be used for:
+        1) When an IO does not support a specific read/write functionality
+        2) A file format does not support the structural requirements
+            * Different sampling rates among streams
+            * Different expectations for a file format (could also be
+            a NotImplementedError depending on circumstances)
+        
+    It should NOT be used when other errors more accurately describe
+    the problem:
+        1) ValueError: for incorrect values (like t_start, t_stop)
+        2) TypeError: use of an inappropriate type for an argument
+        3) FileNotFoundError: for use when a file is not found
+    """
     pass
 
 

neo/io/baseio.py

@@ -35,6 +35,7 @@
     RectangularRegionOfInterest,
     CircularRegionOfInterest,
     PolygonRegionOfInterest,
+    NeoReadWriteError
 )
 
 read_error = "This type is not supported by this file format for reading"
@@ -139,7 +140,7 @@ def read(self, lazy: bool = False, **kargs):
             Returns all the data from the file as Blocks
         """
         if lazy and not self.support_lazy:
-            raise ValueError("This IO module does not support lazy loading")
+            raise NeoReadWriteError("This IO module does not support lazy loading")
         if Block in self.readable_objects:
             if hasattr(self, "read_all_blocks") and callable(getattr(self, "read_all_blocks")):
                 return self.read_all_blocks(lazy=lazy, **kargs)
@@ -168,13 +169,13 @@ def write(self, bl, **kargs):
         if Block in self.writeable_objects:
             if isinstance(bl, Sequence):
                 if not hasattr(self, "write_all_blocks"):
-                    raise AttributeError(f"{self.__class__.__name__} does not offer to store a sequence of blocks")
+                    raise NeoReadWriteError(f"{self.__class__.__name__} does not offer to store a sequence of blocks")
                 self.write_all_blocks(bl, **kargs)
             else:
                 self.write_block(bl, **kargs)
         elif Segment in self.writeable_objects:
             if len(bl.segments) != 1:
-                raise ValueError(f"{self.__class__.__name__} is based on segment so if you try to write a block it "
+                raise NeoReadWriteError(f"{self.__class__.__name__} is based on segment so if you try to write a block it "
                 + "must contain only one Segment")
             self.write_segment(bl.segments[0], **kargs)
         else:
@@ -183,105 +184,105 @@ def write(self, bl, **kargs):
     ######## All individual read methods #######################
     def read_block(self, **kargs):
         if Block not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_segment(self, **kargs):
         if Segment not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_spiketrain(self, **kargs):
         if SpikeTrain not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_analogsignal(self, **kargs):
         if AnalogSignal not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_imagesequence(self, **kargs):
         if ImageSequence not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_rectangularregionofinterest(self, **kargs):
         if RectangularRegionOfInterest not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_circularregionofinterest(self, **kargs):
         if CircularRegionOfInterest not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_polygonregionofinterest(self, **kargs):
         if PolygonRegionOfInterest not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_irregularlysampledsignal(self, **kargs):
         if IrregularlySampledSignal not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_channelview(self, **kargs):
         if ChannelView not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_event(self, **kargs):
         if Event not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_epoch(self, **kargs):
         if Epoch not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     def read_group(self, **kargs):
         if Group not in self.readable_objects:
-            raise RuntimeError(read_error)
+            raise NeoReadWriteError(read_error)
 
     ######## All individual write methods #######################
     def write_block(self, bl, **kargs):
         if Block not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_segment(self, seg, **kargs):
         if Segment not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_spiketrain(self, sptr, **kargs):
         if SpikeTrain not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_analogsignal(self, anasig, **kargs):
         if AnalogSignal not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_imagesequence(self, imseq, **kargs):
         if ImageSequence not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_rectangularregionofinterest(self, rectroi, **kargs):
         if RectangularRegionOfInterest not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_circularregionofinterest(self, circroi, **kargs):
         if CircularRegionOfInterest not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_polygonregionofinterest(self, polyroi, **kargs):
         if PolygonRegionOfInterest not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_irregularlysampledsignal(self, irsig, **kargs):
         if IrregularlySampledSignal not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_channelview(self, chv, **kargs):
         if ChannelView not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_event(self, ev, **kargs):
         if Event not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_epoch(self, ep, **kargs):
         if Epoch not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)
 
     def write_group(self, group, **kargs):
         if Group not in self.writeable_objects:
-            raise RuntimeError(write_error)
+            raise NeoReadWriteError(write_error)

neo/io/igorproio.py

@@ -15,7 +15,7 @@
 import pathlib
 import quantities as pq
 from neo.io.baseio import BaseIO
-from neo.core import Block, Segment, AnalogSignal
+from neo.core import Block, Segment, AnalogSignal, NeoReadWriteError
 
 
 class IgorIO(BaseIO):
@@ -59,14 +59,16 @@ def __init__(self, filename=None, parse_notes=None):
         """
         BaseIO.__init__(self)
         filename = pathlib.Path(filename)
-        assert filename.suffix[1:] in self.extensions, f"Only the following extensions are supported: {self.extensions}"
+        if filename.suffix[1:] not in self.extensions:
+            raise NotImplementedError(f"Only the following extensions are supported: {self.extensions}")
         self.filename = filename
         self.extension = filename.suffix[1:]
         self.parse_notes = parse_notes
         self._filesystem = None
 
     def read_block(self, lazy=False):
-        assert not lazy, "This IO does not support lazy mode"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy reading")
 
         block = Block(file_origin=str(self.filename))
         block.segments.append(self.read_segment(lazy=lazy))
@@ -76,7 +78,8 @@ def read_segment(self, lazy=False):
         import igor2.packed as pxp
         from igor2.record.wave import WaveRecord
 
-        assert not lazy, "This IO does not support lazy mode"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy mode")
         segment = Segment(file_origin=str(self.filename))
 
         if self.extension == "pxp":
@@ -97,15 +100,17 @@ def read_analogsignal(self, path=None, lazy=False):
         import igor2.binarywave as bw
         import igor2.packed as pxp
 
-        assert not lazy, "This IO does not support lazy mode"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy mode")
 
         if self.extension == "ibw":
             data = bw.load(str(self.filename))
             version = data["version"]
             if version > 5:
                 raise IOError(f"Igor binary wave file format version {version} " "is not supported.")
         elif self.extension == "pxp":
-            assert type(path) is str, "A colon-separated Igor-style path must be provided."
+            if type(path) is not str:
+                raise TypeError("A colon-separated Igor-style path must be provided.")
             if not self._filesystem:
                 _, self.filesystem = pxp.load(str(self.filename))
                 path = path.split(":")
@@ -119,24 +124,23 @@ def read_analogsignal(self, path=None, lazy=False):
 
     def _wave_to_analogsignal(self, content, dirpath):
         if "padding" in content:
-            assert content["padding"].size == 0, "Cannot handle non-empty padding"
+            if content["padding"].size != 0:
+                raise NeoReadWriteError("Cannot handle non-empty padding")
         signal = content["wData"]
         note = content["note"]
         header = content["wave_header"]
         name = str(header["bname"].decode("utf-8"))
         units = "".join([x.decode() for x in header["dataUnits"]])
-        try:
-            time_units = "".join([x.decode() for x in header["xUnits"]])
-            assert len(time_units)
-        except:
+        time_units = "".join([x.decode() for x in header["xUnits"]])
+        if len(time_units) == 0:
             time_units = "s"
         try:
             t_start = pq.Quantity(header["hsB"], time_units)
         except KeyError:
             t_start = pq.Quantity(header["sfB"][0], time_units)
         try:
             sampling_period = pq.Quantity(header["hsA"], time_units)
-        except:
+        except KeyError:
             sampling_period = pq.Quantity(header["sfA"][0], time_units)
         if self.parse_notes:
             try:

neo/io/rawbinarysignalio.py

@@ -16,7 +16,7 @@
 import quantities as pq
 
 from neo.io.baseio import BaseIO
-from neo.core import Segment, AnalogSignal
+from neo.core import Segment, AnalogSignal, NeoReadWriteError
 
 from neo.io.basefromrawio import BaseFromRaw
 from neo.rawio.rawbinarysignalrawio import RawBinarySignalRawIO
@@ -88,19 +88,24 @@ def write_segment(self, segment):
             raise NotImplementedError("bytesoffset values other than 0 " + "not supported")
 
         anasigs = segment.analogsignals
-        assert len(anasigs) > 0, "No AnalogSignal"
+        if len(anasigs) == 0:
+            raise NeoReadWriteError("No AnalogSignal to write")
 
         anasig0 = anasigs[0]
         if len(anasigs) == 1 and anasig0.ndim == 2:
             numpy_sigs = anasig0.magnitude
         else:
 
-            assert anasig0.ndim == 1 or (anasig0.ndim == 2 and anasig0.shape[1] == 1)
+            if anasig0.ndim != 1 and (anasig0.ndim == 2 and anasig0.shape[1] != 1):
+                raise NeoReadWriteError("Incorrect analogsignal shaping for write")
             # all AnaologSignal from Segment must have the same length/sampling_rate/dtype
             for anasig in anasigs[1:]:
-                assert anasig.shape == anasig0.shape
-                assert anasig.sampling_rate == anasig0.sampling_rate
-                assert anasig.dtype == anasig0.dtype
+                if anasig.shape != anasig0.shape:
+                    raise ValueError("The shape of one of the analog signals does not match")
+                if anasig.sampling_rate != anasig0.sampling_rate:
+                    raise ValueError("The sampling_rate of one of the analog signals does not match")
+                if anasig.dtype != anasig0.dtype:
+                    raise ValueError("The dtype of one the analog signals does not match")
 
             numpy_sigs = np.empty((anasig0.size, len(anasigs)))
             for i, anasig in enumerate(anasigs):

neo/rawio/micromedrawio.py

@@ -15,6 +15,8 @@
     _event_channel_dtype,
 )
 
+from neo.core import NeoReadWriteError
+
 import numpy as np
 
 import datetime
@@ -64,7 +66,8 @@ def _parse_header(self):
 
             # header version
             (header_version,) = f.read_f("b", offset=175)
-            assert header_version == 4
+            if header_version != 4:
+                raise NotImplementedError(f"`header_version {header_version} is not implemented in neo yet")
 
             # area
             f.seek(176)
@@ -89,7 +92,8 @@ def _parse_header(self):
             for zname in zone_names:
                 zname2, pos, length = f.read_f("8sII")
                 zones[zname] = zname2, pos, length
-                assert zname == zname2.decode("ascii").strip(" ")
+                if zname != zname2.decode("ascii").strip(" "):
+                    raise NeoReadWriteError("expected the zone name to match")
 
             # raw signals memmap
             sig_dtype = "u" + str(Bytes)
@@ -131,7 +135,8 @@ def _parse_header(self):
 
             signal_streams = np.array([("Signals", "0")], dtype=_signal_stream_dtype)
 
-            assert np.unique(signal_channels["sampling_rate"]).size == 1
+            if np.unique(signal_channels["sampling_rate"]).size != 1:
+                raise NeoReadWriteError("The sampling rates must be the same across signal channels")
             self._sampling_rate = float(np.unique(signal_channels["sampling_rate"])[0])
 
             # Event channels
@@ -201,11 +206,13 @@ def _segment_t_stop(self, block_index, seg_index):
         return t_stop
 
     def _get_signal_size(self, block_index, seg_index, stream_index):
-        assert stream_index == 0
+        if stream_index != 0:
+            raise ValueError("`stream_index` must be 0")
         return self._raw_signals.shape[0]
 
     def _get_signal_t_start(self, block_index, seg_index, stream_index):
-        assert stream_index == 0
+        if stream_index != 0:
+            raise ValueError("`stream_index` must be 0")
         return 0.0
 
     def _get_analogsignal_chunk(self, block_index, seg_index, i_start, i_stop, stream_index, channel_indexes):