assert -> errors

Author: zm711

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

neo/io/elphyio.py

@@ -87,7 +87,7 @@
 from neo.io.baseio import BaseIO
 
 # to import from core
-from neo.core import Block, Segment, AnalogSignal, Event, SpikeTrain
+from neo.core import Block, Segment, AnalogSignal, Event, SpikeTrain, NeoReadWriteError
 
 
 # --------------------------------------------------------
@@ -328,7 +328,8 @@ def __init__(
         super().__init__(layout, episode, number, x_unit, n_events, name)
         self.wf_samples = wf_samples
         self.wf_sampling_frequency = wf_sampling_frequency
-        assert wf_sampling_frequency, "bad sampling frequency"
+        if not wf_sampling_frequency:
+            raise ValueError("bad sampling frequency")
         self.wf_sampling_period = 1.0 / wf_sampling_frequency
         self.wf_units = [unit_x_wf, unit_y_wf]
         self.t_start = t_start
@@ -532,7 +533,8 @@ def detect_protocol_from_name(self, path):
         match = re.search(pattern, path)
         if hasattr(match, "end"):
             code = codes.get(path[match.end() - 1].lower(), None)
-            assert code != "m", "multistim file detected"
+            if code == "m":
+                raise ValueError("multistim file detected")
             return code
         elif "spt" in filename.lower():
             return "spontaneousactivity"
@@ -897,7 +899,8 @@ def __init__(self, layout):
 
         # extract episode properties
         n_channels = read_from_char(fileobj, "B")
-        assert not ((n_channels < 1) or (n_channels > 16)), "bad number of channels"
+        if (n_channels < 1) or (n_channels > 16):
+            raise ValueError(f"`n_channels` must be between 1 and 15. It is currently {n_channels}")
         nbpt = read_from_char(fileobj, "h")
         l_xu, x_unit = struct.unpack("<B3s", fileobj.read(4))
         if hasattr(x_unit, "decode"):
@@ -1001,7 +1004,8 @@ def __init__(self, layout):
 
         n_ep = read_from_char(fileobj, "h")
         tpData = read_from_char(fileobj, "h")
-        assert tpData in [3, 2, 1, 0], "bad sample size"
+        if tpData not in [3, 2, 1, 0]:
+            raise ValueError(f"`tpData` must be 3, 2, 1, or 0, but is {tpData}")
         no_analog_data = read_from_char(fileobj, "?")
 
         self.n_ep = n_ep
@@ -1784,8 +1788,10 @@ def get_signal(self, episode, channel):
         Return the signal description relative
         to the specified episode and channel.
         """
-        assert episode in range(1, self.n_episodes + 1)
-        assert channel in range(1, self.n_channels(episode) + 1)
+        if episode not in range(1, self.n_episodes + 1):
+            raise ValueError(f"`episode must be in {range(1, self.n_episodes + 1)}, but is {episode}")
+        if channel not in range(1, self.n_channels(episode) + 1):
+            raise ValueError(f"`channel` must be in {range(1, self.n_channels(episode)+1)} but is {channel}")
         t_start = 0
         sampling_period = self.sampling_period(episode, channel)
         t_stop = sampling_period * self.n_samples(episode, channel)
@@ -1863,7 +1869,8 @@ def reshape_bytes(self, databytes, reshape, datatypes, order="<"):
         """
         Reshape a numpy array containing a set of databytes.
         """
-        assert datatypes and len(datatypes) == len(reshape), "datatypes are not well defined"
+        if not datatypes and len(datatypes) == len(reshape):
+            raise NeoReadWriteError("The `datatypes` are not well-defined")
 
         l_bytes = len(databytes)
 
@@ -2080,7 +2087,8 @@ def get_episode_blocks(self):
 
     def set_info_block(self):
         i_blks = self.get_blocks_of_type("USER INFO")
-        assert len(i_blks) < 2, "too many info blocks"
+        if len(i_blks) >= 2:
+            raise ValueError(f"There are too many info blocks: {i_blks}")
         if len(i_blks):
             self.info_block = i_blks[0]
 
@@ -2205,7 +2213,8 @@ def set_episode_blocks(self):
 
     def set_info_block(self):
         i_blks = self.get_blocks_of_type("USER INFO")
-        assert len(i_blks) < 2, "too many info blocks"
+        if len(i_blks) >= 2:
+            raise ValueError(f"There are too many info blocks: {i_blks}")
         if len(i_blks):
             self.info_block = i_blks[0]
 
@@ -2243,7 +2252,8 @@ def sample_type(self, ep, ch):
 
     def sample_size(self, ep, ch):
         size = super().sample_size(ep, ch)
-        assert size == 2, "sample size is always 2 bytes for DAC2/GS/2000 format"
+        if size != 2:
+            raise ValueError(f"sample size is always 2 bytes for DAC2/GS/2000 format, but size is calculated as {size}")
         return size
 
     def sampling_period(self, ep, ch):
@@ -2297,10 +2307,12 @@ def file_duration(self):
         return self.main_block.dX * self.n_samples
 
     def get_tag(self, episode, tag_channel):
-        assert episode in range(1, self.n_episodes + 1)
+        if episode not in range(1, self.n_episodes + 1):
+            raise ValueError(f"`episode` must be within {range(1 + self.n_episodes + 1)} and is {episode}")
         # there are none or 2 tag channels
         if self.tag_mode(episode) == 1:
-            assert tag_channel in range(1, 3), "DAC2/GS/2000 format support only 2 tag channels"
+            if tag_channel not in range(1, 3):
+                raise ValueError("DAC2/GS/2000 format support only 2 tag channels")
             block = self.episode_block(episode)
             t_stop = self.main_block.n_samples * block.dX
             return ElphyTag(self, episode, tag_channel, block.x_unit, 1.0 / block.dX, 0, t_stop)
@@ -2384,7 +2396,8 @@ def set_episode_blocks(self):
     def set_info_block(self):
         # in fact the file info are contained into a single sub-block with an USR identifier
         i_blks = self.get_blocks_of_type("B_Finfo")
-        assert len(i_blks) < 2, "too many info blocks"
+        if len(i_blks) >= 2:
+            raise ValueError(f"There are too many info blocks: {i_blks}")
         if len(i_blks):
             i_blk = i_blks[0]
             sub_blocks = i_blk.sub_blocks
@@ -2658,7 +2671,8 @@ def get_tag(self, episode, tag_channel):
         Return a :class:`ElphyTag` which is a
         descriptor of the specified event channel.
         """
-        assert episode in range(1, self.n_episodes + 1)
+        if episode not in range(1, self.n_episodes + 1):
+            raise ValueError(f"`episode` must be in {range(1, self.n_episodes + 1)} and is {episode}")
 
         # there are none, 2 or 16 tag
         # channels depending on tag_mode
@@ -2669,11 +2683,11 @@ def get_tag(self, episode, tag_channel):
 
             # verify the validity of the tag channel
             if tag_mode == 1:
-                assert tag_channel in range(1, 3), "Elphy format support only 2 tag channels for tag_mode == 1"
-            elif tag_mode == 2:
-                assert tag_channel in range(1, 17), "Elphy format support only 16 tag channels for tag_mode == 2"
-            elif tag_mode == 3:
-                assert tag_channel in range(1, 17), "Elphy format support only 16 tag channels for tag_mode == 3"
+                if tag_channel not in range(1, 3):
+                    raise ValueError("Elphy format support only 2 tag channels for tag_mode == 1")
+            elif tag_mode == 2 or tag_mode ==3:
+                if tag_channel not in range(1, 17):
+                    raise ValueError("Elphy format support only 16 tag channels for tag_mode == 2 or tag_mode == 3")
 
             smp_period = block.ep_block.dX
             smp_freq = 1.0 / smp_period
@@ -2707,12 +2721,15 @@ def get_event(self, ep, ch, marked_ks):
         Return a :class:`ElphyEvent` which is a
         descriptor of the specified event channel.
         """
-        assert ep in range(1, self.n_episodes + 1)
-        assert ch in range(1, self.n_channels + 1)
+        if ep not in range(1, self.n_episodes + 1):
+            raise ValueError(f"`ep` must be in {range(1, self.n_episodes + 1)}, but is {ep}")
+        if ch not in range(1, self.n_channels + 1):
+            raise ValueError(f"`ch` must be in {range(1, self.n_channels + 1)}, but is {ch}")
 
         # find the event channel number
         evt_channel = np.where(marked_ks == -1)[0][0]
-        assert evt_channel in range(1, self.n_events(ep) + 1)
+        if evt_channel not in range(1, self.n_events(ep) + 1):
+            raise ValueError(f"`evt_channel` must be in {range(1, self.n_events(ep) + 1)}, but is {evt_channel}")
 
         block = self.episode_block(ep)
         ep_blocks = self.get_blocks_stored_in_episode(ep)
@@ -2792,8 +2809,10 @@ def get_spiketrain(self, episode, electrode_id):
         Return a :class:`Spike` which is a
         descriptor of the specified spike channel.
         """
-        assert episode in range(1, self.n_episodes + 1)
-        assert electrode_id in range(1, self.n_spiketrains(episode) + 1)
+        if episode not in range(1, self.n_episodes + 1):
+            raise ValueError(f"`episode` must be in {range(1, self.n_episodes + 1)}, but is {episode}")
+        if electrode_id not in range(1, self.n_spiketrains(episode) + 1):
+            raise ValueError(f"`eletrodee_id` must be in {range(1, self.n_spiketrains(episode)+1)}, but is {electrode_id}")
         # get some properties stored in the episode sub-block
         block = self.episode_block(episode)
         x_unit = block.ep_block.x_unit
@@ -3748,7 +3767,8 @@ def read_block(
         lazy : bool
             Postpone actual reading of the file.
         """
-        assert not lazy, "Do not support lazy"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy reading")
 
         # basic
         block = Block(name=None)

neo/io/nixio.py

@@ -313,7 +313,8 @@ def _nix_to_neo_block(self, nix_block):
                 neo_block.segments.append(newseg)
             elif grp.type == "neo.group":
                 newgrp, parent_name = self._nix_to_neo_group(grp)
-                assert parent_name is None
+                if parent_name is not None:
+                    raise ValueError(f"`parent_name` must be None and is {parent_name}")
                 neo_block.groups.append(newgrp)
             elif grp.type == "neo.subgroup":
                 newgrp, parent_name = self._nix_to_neo_group(grp)

neo/io/nwbio.py

@@ -86,7 +86,10 @@ def get_class(module, name):
     import pynwb
 
     module_path = module.split(".")
-    assert len(module_path) == 2  # todo: handle the general case where this isn't 2
+    if len(module_path) != 2:
+        raise ValueError(
+            f"`module_path` must be 2, not {module_path}"
+        )  # todo: handle the general case where this isn't 2
     return getattr(getattr(pynwb, module_path[1]), name)
 
 
@@ -139,8 +142,10 @@ def _decompose_unit(unit):
     >>> _decompose_unit(pq.mV)
     ('volt', 0.001)
     """
-    assert isinstance(unit, pq.quantity.Quantity)
-    assert unit.magnitude == 1
+    if not isinstance(unit, pq.quantity.Quantity):
+        raise TypeError(f"`unit` must be of type pq.quantity.Quantity and not type {type(unit)}")
+    if unit.magnitude != 1:
+        raise ValueError(f"The magnitude of the `unit` must be 1 not {unit.magnitude}")
     conversion = 1.0
 
     def _decompose(unit):
@@ -240,7 +245,8 @@ def read_all_blocks(self, lazy=False, **kwargs):
         """
         import pynwb
 
-        assert self.nwb_file_mode in ("r",)
+        if self.nwb_file_mode not in ("r",):
+            raise ValueError("`mode` at init needs to be set to 'r' to read files")
         self._io_nwb = pynwb.NWBHDF5IO(
             self.filename, mode=self.nwb_file_mode, load_namespaces=True
         )  # Open a file with NWBHDF5IO
@@ -320,7 +326,8 @@ def _read_epochs_group(self, lazy):
                         epoch = epoch.load()
                     segment_name = np.unique(segment_names[index])
                     block_name = np.unique(block_names[index])
-                    assert segment_name.size == block_name.size == 1
+                    if segment_name.size != block_name.size == 1:
+                        raise ValueError("the `segment_name` and the `block_name` should be the same")
                     segment = self._get_segment(block_name[0], segment_name[0])
                     segment.epochs.append(epoch)
             else:
@@ -449,7 +456,8 @@ def write_all_blocks(self, blocks, validate=True, **kwargs):
         for i, block in enumerate(blocks):
             self._write_block(block)
 
-        assert self.nwb_file_mode in ("w",)  # possibly expand to 'a'ppend later
+        if self.nwb_file_mode not in ("w",):
+            raise ValueError("mode must be 'w' in order to write files")  # possibly expand to 'a'ppend later
         if self.nwb_file_mode == "w" and os.path.exists(self.filename):
             os.remove(self.filename)
         io_nwb = pynwb.NWBHDF5IO(self.filename, mode=self.nwb_file_mode)
@@ -483,7 +491,8 @@ def _write_block(self, block):
         if not block.name:
             block.name = f"block{self.blocks_written}"
         for i, segment in enumerate(block.segments):
-            assert segment.block is block
+            if segment.block is not block:
+                raise TypeError(f"segment.block must be block it is {segment.block}")
             if not segment.name:
                 segment.name = f"{block.name} : segment{i}"
             self._write_segment(self._nwbfile, segment, electrodes)
@@ -513,7 +522,8 @@ def _write_electrodes(self, nwbfile, block):
     def _write_segment(self, nwbfile, segment, electrodes):
         # maybe use NWB trials to store Segment metadata?
         for i, signal in enumerate(chain(segment.analogsignals, segment.irregularlysampledsignals)):
-            assert signal.segment is segment
+            if signal.segment is not segment:
+                raise TypeError(f"signal.segment must be segment and is {signal.segment}")
             if hasattr(signal, "name"):
                 signal.name = f"{segment.name} {signal.name} {i}"
                 logging.warning(f"Warning signal name exists. New name: {signal.name}")
@@ -522,13 +532,15 @@ def _write_segment(self, nwbfile, segment, electrodes):
             self._write_signal(self._nwbfile, signal, electrodes)
 
         for i, train in enumerate(segment.spiketrains):
-            assert train.segment is segment
+            if train.segment is not segment:
+                raise TypeError(f"train.segment must be segment and is {train.segment}")
             if not train.name:
                 train.name = f"{segment.name} : spiketrain{i}"
             self._write_spiketrain(self._nwbfile, train)
 
         for i, event in enumerate(segment.events):
-            assert event.segment is segment
+            if event.segment is not segment:
+                raise TypeError(f"event.segment mst be segment and is {event.segment}")
             if hasattr(event, "name"):
                 event.name = f"{segment.name}  {event.name} {i}"
                 logging.warning(f"Warning event name exists. New name: {event.name}")

neo/io/pickleio.py

@@ -13,7 +13,7 @@
 import pickle
 
 from neo.io.baseio import BaseIO
-from neo.core import Block, Segment, AnalogSignal, SpikeTrain
+from neo.core import Block, Segment, AnalogSignal, SpikeTrain, NeoReadWriteError
 
 
 class PickleIO(BaseIO):
@@ -38,7 +38,8 @@ class PickleIO(BaseIO):
     extensions = ["pkl", "pickle"]
 
     def read_block(self, lazy=False):
-        assert not lazy, "Do not support lazy"
+        if lazy:
+            raise NeoReadWriteError("This IO does not support lazy reading")
         with open(self.filename, "rb") as fp:
             block = pickle.load(fp)
         return block