Merge pull request #1452 from zm711/assert-errors

Convert Asserts to Errors

Author: samuelgarcia

neo/core/__init__.py

@@ -51,6 +51,8 @@
 from neo.core.view import ChannelView
 from neo.core.group import Group
 
+from neo.core.baseneo import NeoReadWriteError
+
 # Block should always be first in this list
 objectlist = [
     Block,

neo/core/baseneo.py

@@ -34,6 +34,27 @@
 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
+
 
 def _check_annotations(value):
     """
@@ -68,7 +89,8 @@ def merge_annotation(a, b):
         For strings: concatenate with ';'
         Otherwise: fail if the annotations are not equal
     """
-    assert type(a) == type(b), f"type({a})) {type(a)} != type({b}) {type(b)}"
+    if type(a) != type(b):
+        raise TypeError(f"type({a}) {type(a)} != type({b}) {type(b)}")
     if isinstance(a, dict):
         return merge_annotations(a, b)
     elif isinstance(a, np.ndarray):  # concatenate b to a
@@ -81,7 +103,8 @@ def merge_annotation(a, b):
         else:
             return a + ";" + b
     else:
-        assert a == b, f"{a} != {b}"
+        if a != b:
+            raise ValueError(f"{a} != {b}")
         return a
 
 
@@ -131,7 +154,8 @@ def intersect_annotations(A, B):
 
     for key in set(A.keys()) & set(B.keys()):
         v1, v2 = A[key], B[key]
-        assert type(v1) == type(v2), f"type({v1}) {type(v1)} != type({v2}) {type(v2)}"
+        if type(v1) != type(v2):
+            raise TypeError(f"type({v1}) {type(v1)} != type({v2}) {type(v2)}")
         if isinstance(v1, dict) and v1 == v2:
             result[key] = deepcopy(v1)
         elif isinstance(v1, str) and v1 == v2:

neo/core/container.py

@@ -450,7 +450,8 @@ def check_relationships(self, recursive=True):
             else:
                 container = getattr(self, _container_name(child.__class__.__name__))
             if container.parent is not None:
-                assert getattr(child, parent_name, None) is self
+                if getattr(child, parent_name, None) is not self:
+                    raise AttributeError("Child should have its parent as an attribute")
         if recursive:
             for child in self.container_children:
                 child.check_relationships(recursive=True)

neo/core/objectlist.py

@@ -19,7 +19,8 @@ def __init__(self, allowed_contents, parent=None):
             self.allowed_contents = (allowed_contents,)
         else:
             for item in allowed_contents:
-                assert issubclass(item, BaseNeo)
+                if not issubclass(item, BaseNeo):
+                    raise TypeError("Each item in allowed_contents must be a subclass of BaseNeo")
             self.allowed_contents = tuple(allowed_contents)
         self._items = []
         self.parent = parent

neo/core/spiketrain.py

@@ -891,8 +891,8 @@ def _merge_array_annotations(self, others, sorting=None):
         the spikes.
         :return Merged array_annotations
         """
-
-        assert sorting is not None, "The order of the merged spikes must be known"
+        if sorting is None:
+            raise ValueError("The order of the merged spikes must be known")
 
         merged_array_annotations = {}
 

neo/core/spiketrainlist.py

@@ -87,8 +87,8 @@ def __init__(self, items=None, parent=None):
         self._channel_id_array = None
         self._all_channel_ids = None
         self._spiketrain_metadata = {}
-        if parent is not None:
-            assert parent.__class__.__name__ == "Segment"
+        if parent is not None and parent.__class__.__name__ != "Segment":
+            raise AttributeError("The parent class must be a Segment")
         self.segment = parent
 
     @property

neo/io/asciispiketrainio.py

@@ -14,7 +14,7 @@
 import quantities as pq
 
 from neo.io.baseio import BaseIO
-from neo.core import Segment, SpikeTrain
+from neo.core import Segment, SpikeTrain, NeoReadWriteError
 
 
 class AsciiSpikeTrainIO(BaseIO):
@@ -91,7 +91,8 @@ def read_segment(
             t_start : time start of all spiketrain 0 by default
             unit : unit of spike times, can be a str or directly a Quantities
         """
-        assert not lazy, "Do not support lazy"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy reading")
 
         unit = pq.Quantity(1, unit)
 

neo/io/basefromrawio.py

@@ -241,13 +241,10 @@ def read_segment(
         """
 
         if lazy:
-            assert (
-                time_slice is None
-            ), "For lazy=True you must specify a time_slice when LazyObject.load(time_slice=...)"
-
-            assert (
-                not load_waveforms
-            ), "For lazy=True you must specify load_waveforms when SpikeTrain.load(load_waveforms=...)"
+            if time_slice is not None:
+                raise ValueError("For lazy=True you must specify a time_slice when LazyObject.load(time_slice=...)")
+            if load_waveforms:
+                raise ValueError("For lazy=True you must specify load_waveforms when SpikeTrain.load(load_waveforms=...)")
 
         if signal_group_mode is None:
             signal_group_mode = self._prefered_signal_group_mode

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)
@@ -167,97 +168,121 @@ def write(self, bl, **kargs):
         """
         if Block in self.writeable_objects:
             if isinstance(bl, Sequence):
-                assert hasattr(
-                    self, "write_all_blocks"
-                ), f"{self.__class__.__name__} does not offer to store a sequence of blocks"
+                if not hasattr(self, "write_all_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:
-            assert len(bl.segments) == 1, (
-                f"{self.__class__.__name__} is based on segment so if you try to write a block it "
-                + "must contain only one Segment"
-            )
+            if len(bl.segments) != 1:
+                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:
             raise NotImplementedError
 
     ######## All individual read methods #######################
     def read_block(self, **kargs):
-        assert Block in self.readable_objects, read_error
+        if Block not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_segment(self, **kargs):
-        assert Segment in self.readable_objects, read_error
+        if Segment not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_spiketrain(self, **kargs):
-        assert SpikeTrain in self.readable_objects, read_error
+        if SpikeTrain not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_analogsignal(self, **kargs):
-        assert AnalogSignal in self.readable_objects, read_error
+        if AnalogSignal not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_imagesequence(self, **kargs):
-        assert ImageSequence in self.readable_objects, read_error
+        if ImageSequence not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_rectangularregionofinterest(self, **kargs):
-        assert RectangularRegionOfInterest in self.readable_objects, read_error
+        if RectangularRegionOfInterest not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_circularregionofinterest(self, **kargs):
-        assert CircularRegionOfInterest in self.readable_objects, read_error
+        if CircularRegionOfInterest not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_polygonregionofinterest(self, **kargs):
-        assert PolygonRegionOfInterest in self.readable_objects, read_error
+        if PolygonRegionOfInterest not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_irregularlysampledsignal(self, **kargs):
-        assert IrregularlySampledSignal in self.readable_objects, read_error
+        if IrregularlySampledSignal not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_channelview(self, **kargs):
-        assert ChannelView in self.readable_objects, read_error
+        if ChannelView not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_event(self, **kargs):
-        assert Event in self.readable_objects, read_error
+        if Event not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_epoch(self, **kargs):
-        assert Epoch in self.readable_objects, read_error
+        if Epoch not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     def read_group(self, **kargs):
-        assert Group in self.readable_objects, read_error
+        if Group not in self.readable_objects:
+            raise NeoReadWriteError(read_error)
 
     ######## All individual write methods #######################
     def write_block(self, bl, **kargs):
-        assert Block in self.writeable_objects, write_error
+        if Block not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_segment(self, seg, **kargs):
-        assert Segment in self.writeable_objects, write_error
+        if Segment not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_spiketrain(self, sptr, **kargs):
-        assert SpikeTrain in self.writeable_objects, write_error
+        if SpikeTrain not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_analogsignal(self, anasig, **kargs):
-        assert AnalogSignal in self.writeable_objects, write_error
+        if AnalogSignal not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_imagesequence(self, imseq, **kargs):
-        assert ImageSequence in self.writeable_objects, write_error
+        if ImageSequence not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_rectangularregionofinterest(self, rectroi, **kargs):
-        assert RectangularRegionOfInterest in self.writeable_objects, read_error
+        if RectangularRegionOfInterest not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_circularregionofinterest(self, circroi, **kargs):
-        assert CircularRegionOfInterest in self.writeable_objects, read_error
+        if CircularRegionOfInterest not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_polygonregionofinterest(self, polyroi, **kargs):
-        assert PolygonRegionOfInterest in self.writeable_objects, read_error
+        if PolygonRegionOfInterest not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_irregularlysampledsignal(self, irsig, **kargs):
-        assert IrregularlySampledSignal in self.writeable_objects, write_error
+        if IrregularlySampledSignal not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_channelview(self, chv, **kargs):
-        assert ChannelView in self.writeable_objects, write_error
+        if ChannelView not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_event(self, ev, **kargs):
-        assert Event in self.writeable_objects, write_error
+        if Event not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_epoch(self, ep, **kargs):
-        assert Epoch in self.writeable_objects, write_error
+        if Epoch not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)
 
     def write_group(self, group, **kargs):
-        assert Group in self.writeable_objects, write_error
+        if Group not in self.writeable_objects:
+            raise NeoReadWriteError(write_error)

neo/io/brainwaref32io.py

@@ -32,7 +32,7 @@
 import quantities as pq
 
 # needed core neo modules
-from neo.core import Block, Group, Segment, SpikeTrain
+from neo.core import Block, Group, Segment, SpikeTrain, NeoReadWriteError
 
 # need to subclass BaseIO
 from neo.io.baseio import BaseIO
@@ -127,7 +127,8 @@ def read_block(self, lazy=False, **kargs):
         Reads a block from the simple spike data file "fname" generated
         with BrainWare
         """
-        assert not lazy, "Do not support lazy"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy reading")
 
         # there are no keyargs implemented to so far.  If someone tries to pass
         # them they are expecting them to do something or making a mistake,