convert assert to errors I

Author: zm711

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/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

@@ -167,97 +167,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 AttributeError(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 ValueError(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 RuntimeError(read_error)
 
     def read_segment(self, **kargs):
-        assert Segment in self.readable_objects, read_error
+        if Segment not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_spiketrain(self, **kargs):
-        assert SpikeTrain in self.readable_objects, read_error
+        if SpikeTrain not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_analogsignal(self, **kargs):
-        assert AnalogSignal in self.readable_objects, read_error
+        if AnalogSignal not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_imagesequence(self, **kargs):
-        assert ImageSequence in self.readable_objects, read_error
+        if ImageSequence not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_rectangularregionofinterest(self, **kargs):
-        assert RectangularRegionOfInterest in self.readable_objects, read_error
+        if RectangularRegionOfInterest not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_circularregionofinterest(self, **kargs):
-        assert CircularRegionOfInterest in self.readable_objects, read_error
+        if CircularRegionOfInterest not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_polygonregionofinterest(self, **kargs):
-        assert PolygonRegionOfInterest in self.readable_objects, read_error
+        if PolygonRegionOfInterest not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_irregularlysampledsignal(self, **kargs):
-        assert IrregularlySampledSignal in self.readable_objects, read_error
+        if IrregularlySampledSignal not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_channelview(self, **kargs):
-        assert ChannelView in self.readable_objects, read_error
+        if ChannelView not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_event(self, **kargs):
-        assert Event in self.readable_objects, read_error
+        if Event not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_epoch(self, **kargs):
-        assert Epoch in self.readable_objects, read_error
+        if Epoch not in self.readable_objects:
+            raise RuntimeError(read_error)
 
     def read_group(self, **kargs):
-        assert Group in self.readable_objects, read_error
+        if Group not in self.readable_objects:
+            raise RuntimeError(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 RuntimeError(write_error)
 
     def write_segment(self, seg, **kargs):
-        assert Segment in self.writeable_objects, write_error
+        if Segment not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_spiketrain(self, sptr, **kargs):
-        assert SpikeTrain in self.writeable_objects, write_error
+        if SpikeTrain not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_analogsignal(self, anasig, **kargs):
-        assert AnalogSignal in self.writeable_objects, write_error
+        if AnalogSignal not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_imagesequence(self, imseq, **kargs):
-        assert ImageSequence in self.writeable_objects, write_error
+        if ImageSequence not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_rectangularregionofinterest(self, rectroi, **kargs):
-        assert RectangularRegionOfInterest in self.writeable_objects, read_error
+        if RectangularRegionOfInterest not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_circularregionofinterest(self, circroi, **kargs):
-        assert CircularRegionOfInterest in self.writeable_objects, read_error
+        if CircularRegionOfInterest not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_polygonregionofinterest(self, polyroi, **kargs):
-        assert PolygonRegionOfInterest in self.writeable_objects, read_error
+        if PolygonRegionOfInterest not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_irregularlysampledsignal(self, irsig, **kargs):
-        assert IrregularlySampledSignal in self.writeable_objects, write_error
+        if IrregularlySampledSignal not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_channelview(self, chv, **kargs):
-        assert ChannelView in self.writeable_objects, write_error
+        if ChannelView not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_event(self, ev, **kargs):
-        assert Event in self.writeable_objects, write_error
+        if Event not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_epoch(self, ep, **kargs):
-        assert Epoch in self.writeable_objects, write_error
+        if Epoch not in self.writeable_objects:
+            raise RuntimeError(write_error)
 
     def write_group(self, group, **kargs):
-        assert Group in self.writeable_objects, write_error
+        if Group not in self.writeable_objects:
+            raise RuntimeError(write_error)

neo/rawio/baserawio.py

@@ -497,23 +497,23 @@ def _check_stream_signal_channel_characteristics(self):
         signal_streams = self.header["signal_streams"]
         signal_channels = self.header["signal_channels"]
         if signal_streams.size > 0:
-            assert signal_channels.size > 0, "Signal stream but no signal_channels!!!"
+            if signal_channels.size < 1:
+                raise ValueError("Signal stream exists but there are no signal channels")
+            
 
         for stream_index in range(signal_streams.size):
             stream_id = signal_streams[stream_index]["id"]
             mask = signal_channels["stream_id"] == stream_id
             characteristics = signal_channels[mask][_common_sig_characteristics]
             unique_characteristics = np.unique(characteristics)
-            assert unique_characteristics.size == 1, (
-                f"Some channels in stream_id {stream_id} "
-                f"do not have same {_common_sig_characteristics} {unique_characteristics}"
-            )
+            if unique_characteristics.size != 1:
+                raise ValueError(f"Some channels in stream_id {stream_id} "
+                                 f"do not have the same {_common_sig_characteristics} {unique_characteristics}")
 
             # also check that channel_id is unique inside a stream
             channel_ids = signal_channels[mask]["id"]
-            assert (
-                np.unique(channel_ids).size == channel_ids.size
-            ), f"signal_channels do not have unique ids for stream {stream_index}"
+            if np.unique(channel_ids) != channel_ids.size:
+                raise ValueError(f"signal_channels do not have unique ids for stream {stream_index}")
 
         self._several_channel_groups = signal_streams.size > 1
 
@@ -540,7 +540,8 @@ def channel_name_to_index(self, stream_index: int, channel_names: list[str]):
         mask = self.header["signal_channels"]["stream_id"] == stream_id
         signal_channels = self.header["signal_channels"][mask]
         chan_names = list(signal_channels["name"])
-        assert signal_channels.size == np.unique(chan_names).size, "Channel names not unique"
+        if signal_channels.size != np.unique(chan_names).size:
+            raise ValueError("Channel names are not unique")
         channel_indexes = np.array([chan_names.index(name) for name in channel_names])
         return channel_indexes
 
@@ -621,12 +622,12 @@ def _get_stream_index_from_arg(self, stream_index_arg: int | None):
 
         """
         if stream_index_arg is None:
-            assert self.header["signal_streams"].size == 1, "stream_index must be given for multiple stream files"
+            if self.header["signal_streams"].size != 1:
+                raise ValueError("stream_index must be given for files with multiple streams")
             stream_index = 0
         else:
-            assert (
-                0 <= stream_index_arg < self.header["signal_streams"].size
-            ), f"stream_index must be between 0 and {self.header['signal_streams'].size}"
+            if stream_index_arg < 0 or stream_index_arg >= self.header["signal_streams"].size:
+                raise ValueError(f"stream_index must be between 0 and {self.header['signal_streams'].size}")
             stream_index = stream_index_arg
         return stream_index
 
@@ -786,7 +787,9 @@ def get_analogsignal_chunk(
 
         if isinstance(channel_indexes, np.ndarray):
             if channel_indexes.dtype == "bool":
-                assert self.signal_channels_count(stream_index) == channel_indexes.size
+                if self.signal_channels_count(stream_index) != channel_indexes.size:
+                    raise ValueError("If channel_indexes is a boolean it must have be the same length as the "
+                                     f"number of channels {self.signal_channels_count(stream_index)}")
                 (channel_indexes,) = np.nonzero(channel_indexes)
 
         if prefer_slice and isinstance(channel_indexes, np.ndarray):
@@ -1210,9 +1213,8 @@ def setup_cache(self, cache_path: "home" | "same_as_resource", **init_kargs):
         elif cache_path == "same_as_resource":
             dirname = os.path.dirname(resource_name)
         else:
-            assert os.path.exists(
-                cache_path
-            ), 'cache_path does not exists use "home" or "same_as_resource" to make this auto'
+            if not os.path.exists(cache_path):
+                raise ValueError("cache_path does not exist use 'home' or 'same_as_resource' to make this auto")
 
         # the hash of the resource (dir of file) is done with filename+datetime
         # TODO make something more sophisticated when rawmode='one-dir' that use all
@@ -1233,12 +1235,14 @@ def setup_cache(self, cache_path: "home" | "same_as_resource", **init_kargs):
             self.dump_cache()
 
     def add_in_cache(self, **kargs):
-        assert self.use_cache
+        if not self.use_cache:
+            raise ValueError("Can not use add_in_cache if not using cache")
         self._cache.update(kargs)
         self.dump_cache()
 
     def dump_cache(self):
-        assert self.use_cache
+        if not self.use_cache:
+            raise ValueError("Can not use dump_cache if not using cache")
         joblib.dump(self._cache, self.cache_filename)
 
     ##################
@@ -1336,7 +1340,8 @@ def _rescale_epoch_duration(self, raw_duration: np.ndarray, dtype: np.dtype):
 
 def pprint_vector(vector, lim: int = 8):
     vector = np.asarray(vector)
-    assert vector.ndim == 1
+    if vector.ndim != 1:
+        raise ValueError(f"`vector` must have a dimension of 1 and not {vector.ndim}")
     if len(vector) > lim:
         part1 = ", ".join(e for e in vector[: lim // 2])
         part2 = " , ".join(e for e in vector[-lim // 2 :])