add neoreadwriteerror + more asserts

Author: zm711

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,9 @@
 class MergeError(Exception):
     pass
 
+class NeoReadWriteError(IOError):
+    pass
+
 
 def _check_annotations(value):
     """
@@ -68,7 +71,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 +85,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 +136,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/io/neomatlabio.py

@@ -39,6 +39,7 @@
     PolygonRegionOfInterest,
     objectnames,
     class_by_name,
+    NeoReadWriteError,
 )
 from neo.core.regionofinterest import RegionOfInterest
 from neo.core.baseneo import _container_name
@@ -262,7 +263,8 @@ def read_block(self, lazy=False):
         """
         import scipy.io
 
-        assert not lazy, "Does not support lazy"
+        if lazy:
+            raise NeoReadWriteError(f"This IO does not support lazy reading")
 
         d = scipy.io.loadmat(self.filename, struct_as_record=False, squeeze_me=True, mat_dtype=True)
         if "block" not in d:
@@ -420,7 +422,8 @@ def create_ob_from_struct(self, struct, classname):
                 value = getattr(struct, attr[0])
                 if i == 0:
                     # this is a bit hacky, should really add an attribute _view_attr to ChannelView and RegionOfInterest
-                    assert isinstance(value, int)  # object id
+                    if not isinstance(value, int):  # object id
+                        raise TypeError(f"value must be int not of type {type(value)}")
                     kwargs[attr[0]] = _Ref(identifier=value, target_class_name=struct.viewed_classname)
                 else:
                     if attr[1] == np.ndarray and isinstance(value, int):
@@ -531,13 +534,18 @@ def _resolve_references(self, bl):
                     container = getattr(grp, container_name)
                     for i, item in enumerate(container):
                         if isinstance(item, _Ref):
-                            assert isinstance(item.identifier, (int, np.integer))
+                            if not isinstance(item.identifier, (int, np.integer)):
+                                raise TypeError(
+                                    f"item.identifier must be either int or np.integer not of type {type(item.identifier)}"
+                                )
                             # A reference to an object that already exists
                             container[i] = obj_lookup[item.identifier]
                         else:
                             # ChannelView and RegionOfInterest
-                            assert item.is_view
-                            assert isinstance(item.obj, _Ref)
+                            if not item.is_view:
+                                raise TypeError(f"`item` must be a view")
+                            if not isinstance(item.obj, _Ref):
+                                raise TypeError(f"`item.obj` must be a {_Ref} and is of type {type(item.obj)}")
                             item.obj = obj_lookup[item.obj.identifier]
 
 

neo/rawio/axonrawio.py

@@ -53,6 +53,7 @@
     _spike_channel_dtype,
     _event_channel_dtype,
 )
+from neo.core import NeoReadWriteError
 
 import numpy as np
 
@@ -124,7 +125,8 @@ def _parse_header(self):
         elif version >= 2.0:
             mode = info["protocol"]["nOperationMode"]
 
-        assert mode in [1, 2, 3, 5], f"Mode {mode} is not supported"
+        if mode not in [1, 2, 3, 5]:
+            raise NeoReadWriteError(f"Mode {mode} is not currently supported in Neo")
         # event-driven variable-length mode (mode 1)
         # event-driven fixed-length mode (mode 2 or 5)
         # gap free mode (mode 3) can be in several episodes

neo/rawio/blackrockrawio.py

@@ -73,6 +73,8 @@
     _event_channel_dtype,
 )
 
+from neo.core import NeoReadWriteError
+
 
 class BlackrockRawIO(BaseRawIO):
     """
@@ -330,15 +332,15 @@ def _parse_header(self):
         else:
             raise (ValueError("nsx_to_load is wrong"))
 
-        assert all(
-            nsx_nb in self._avail_nsx for nsx_nb in self.nsx_to_load
-        ), "nsx_to_load do not match available nsx list"
+        if not all(nsx_nb in self._avail_nsx for nsx_nb in self.nsx_to_load):
+            raise FileNotFoundError(f"nsx_to_load does not match available nsx list")
 
         # check that all files come from the same specification
         all_spec = [self.__nsx_spec[nsx_nb] for nsx_nb in self.nsx_to_load]
         if self._avail_files["nev"]:
             all_spec.append(self.__nev_spec)
-        assert all(all_spec[0] == spec for spec in all_spec), "Files don't have the same internal version"
+        if not all(all_spec[0] == spec for spec in all_spec):
+            raise NeoReadWriteError("Files don't have the same internal version")
 
         if len(self.nsx_to_load) > 0 and self.__nsx_spec[self.nsx_to_load[0]] == "2.1" and not self._avail_files["nev"]:
             pass
@@ -401,9 +403,8 @@ def _parse_header(self):
 
             # check nb segment per nsx
             nb_segments_for_nsx = [len(self.nsx_datas[nsx_nb]) for nsx_nb in self.nsx_to_load]
-            assert all(
-                nb == nb_segments_for_nsx[0] for nb in nb_segments_for_nsx
-            ), "Segment nb not consistent across nsX files"
+            if not all(nb == nb_segments_for_nsx[0] for nb in nb_segments_for_nsx):
+                raise NeoReadWriteError("Segment nb not consistent across nsX files")
             self._nb_segment = nb_segments_for_nsx[0]
 
             self.__delete_empty_segments()
@@ -1263,11 +1264,12 @@ def __match_nsx_and_nev_segment_ids(self, nsx_nb):
                         ev_ids[mask_after_seg] += 1
 
             # consistency check: same number of segments for nsx and nev data
-            assert nb_possible_nev_segments == len(nonempty_nsx_segments), (
-                f"Inconsistent ns{nsx_nb} and nev file. {nb_possible_nev_segments} "
-                f"segments present in .nev file, but {len(nonempty_nsx_segments)} in "
-                "ns{nsx_nb} file."
-            )
+            if nb_possible_nev_segments != len(nonempty_nsx_segments):
+                raise NeoReadWriteError(
+                    f"Inconsistent ns{nsx_nb} and nev file. {nb_possible_nev_segments} "
+                    f"segments present in .nev file, but {len(nonempty_nsx_segments)} in "
+                    f"ns{nsx_nb} file."
+                )
 
             new_nev_segment_id_mapping = dict(zip(range(nb_possible_nev_segments), sorted(list(nonempty_nsx_segments))))
 

neo/rawio/intanrawio.py

@@ -24,6 +24,7 @@
 from packaging.version import Version as V
 
 import numpy as np
+from neo.core import NeoReadWriteError
 
 from .baserawio import (
     BaseRawIO,
@@ -166,9 +167,10 @@ def _parse_header(self):
             time_stream_index = max(self._raw_data.keys())
             timestamp = self._raw_data[time_stream_index][0]
 
-        assert np.all(np.diff(timestamp) == 1), (
-            "Timestamp have gaps, this could be due " "to a corrupted file or an inappropriate file merge"
-        )
+        if not np.all(np.diff(timestamp) == 1):
+            raise NeoReadWriteError(
+                f"Timestamp have gaps, this could be due to a corrupted file or an inappropriate file merge"
+            )
 
         # signals
         signal_channels = []
@@ -448,7 +450,8 @@ def read_rhs(filename):
             if bool(group_info["signal_group_enabled"]):
                 for c in range(group_info["channel_num"]):
                     chan_info = read_variable_header(f, rhs_signal_channel_header)
-                    assert chan_info["signal_type"] not in (1, 2)
+                    if chan_info["signal_type"] in (1, 2):
+                        raise NeoReadWriteError("signal_type of 1 or 2 is not yet implemented in Neo")
                     if bool(chan_info["channel_enabled"]):
                         channels_by_type[chan_info["signal_type"]].append(chan_info)
 

neo/rawio/maxwellrawio.py

@@ -31,6 +31,8 @@
     _event_channel_dtype,
 )
 
+from neo.core import NeoReadWriteError
+
 import numpy as np
 
 
@@ -94,7 +96,8 @@ def _parse_header(self):
                         f"Possible rec_names: {unique_rec_names}"
                     )
                 else:
-                    assert self.rec_name in unique_rec_names, f"rec_name {self.rec_name} not found"
+                    if self.rec_name not in unique_rec_names:
+                        raise NeoReadWriteError(f"rec_name {self.rec_name} not found")
             else:
                 self.rec_name = unique_rec_names[0]
             # add streams that contain the selected rec_name

neo/rawio/openephysrawio.py

@@ -21,6 +21,8 @@
     _event_channel_dtype,
 )
 
+from neo.core import NeoReadWriteError
+
 
 RECORD_SIZE = 1024
 HEADER_SIZE = 1024
@@ -143,9 +145,8 @@ def _parse_header(self):
 
                 if channel_has_gaps:
                     # protect against strange timestamp block like in file 'OpenEphys_SampleData_3' CH32
-                    assert (
-                        np.median(diff) == RECORD_SIZE
-                    ), f"This file has a non valid data block size for channel {chan_id}, this case cannot be handled"
+                    if not np.median(diff) == RECORD_SIZE:
+                        raise NeoReadWriteError(f"This file has a non valid data block size for channel {chan_id}, this case cannot be handled")
 
                 if seg_index == 0:
                     # add in channel list
@@ -195,7 +196,8 @@ def _parse_header(self):
                 last = all_last_timestamps[0]
 
             # check unique sampling rate
-            assert all(all_samplerate[0] == e for e in all_samplerate), "Not all signals have the same sample rate"
+            if not all(all_samplerate[0] == e for e in all_samplerate):
+                raise NeoReadWriteError("Not all signals have the same sample rate")
 
             self._sig_length[seg_index] = last - first
             self._sig_timestamp0[seg_index] = first
@@ -250,7 +252,8 @@ def _parse_header(self):
                 if self._spike_sampling_rate is None:
                     self._spike_sampling_rate = spike_info["sampleRate"]
                 else:
-                    assert self._spike_sampling_rate == spike_info["sampleRate"], "mismatch in spike sampling rate"
+                    if self._spike_sampling_rate != spike_info["sampleRate"]:
+                        raise ValueError("There is a mismatch in spike sampling rate")
 
                 # scan all to detect several all unique(sorted_ids)
                 all_sorted_ids = []

neo/rawio/phyrawio.py

@@ -197,11 +197,13 @@ def _parse_header(self):
                 spiketrain_an["pc_feature_ind"] = self._pc_feature_ind[unique_templates]
 
     def _segment_t_start(self, block_index, seg_index):
-        assert block_index == 0
+        if block_index != 0:
+            raise ValueError("`block_index` must be 0")
         return self._t_start
 
     def _segment_t_stop(self, block_index, seg_index):
-        assert block_index == 0
+        if block_index != 0:
+            raise ValueError("`block_index` must be 0")
         return self._t_stop
 
     def _get_signal_size(self, block_index, seg_index, channel_indexes=None):
@@ -214,16 +216,19 @@ def _get_analogsignal_chunk(self, block_index, seg_index, i_start, i_stop, chann
         return None
 
     def _spike_count(self, block_index, seg_index, spike_channel_index):
-        assert block_index == 0
+        if block_index != 0:
+            raise ValueError("`block_index` must be 0")
         spikes = self._spike_clusters
         unit_label = self.unit_labels[spike_channel_index]
         mask = spikes == unit_label
         nb_spikes = np.sum(mask)
         return nb_spikes
 
     def _get_spike_timestamps(self, block_index, seg_index, spike_channel_index, t_start, t_stop):
-        assert block_index == 0
-        assert seg_index == 0
+        if block_index != 0:
+            raise ValueError("`block_index` must be 0")
+        if seg_index != 0:
+            raise ValueError("`seg_index` must be 0")
 
         unit_label = self.unit_labels[spike_channel_index]
         mask = self._spike_clusters == unit_label

neo/rawio/rawbinarysignalrawio.py

@@ -143,11 +143,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):