diff --git a/neo/core/container.py b/neo/core/container.py index 8f9980ad1..77d12a360 100644 --- a/neo/core/container.py +++ b/neo/core/container.py @@ -226,7 +226,7 @@ def _set_object_list(self, name, value): elif isinstance(value, ObjectList): # from __iadd__ setattr(self, name, value) else: - TypeError("value must be a list or an ObjectList") + raise TypeError("value must be a list or an ObjectList") @property def _child_objects(self): diff --git a/neo/core/objectlist.py b/neo/core/objectlist.py index f3209cda5..2169ab936 100644 --- a/neo/core/objectlist.py +++ b/neo/core/objectlist.py @@ -27,6 +27,12 @@ def __init__(self, allowed_contents, parent=None): self.parent = parent def _handle_append(self, obj): + self._check_allowed(obj) + if self._contains(obj): + raise ValueError("Cannot add this object because it is already contained within the list") + self._set_parent(obj) + + def _check_allowed(self, obj): if not ( isinstance(obj, self.allowed_contents) or ( # also allow proxy objects of the correct type @@ -35,9 +41,7 @@ def _handle_append(self, obj): ): raise TypeError(f"Object is a {type(obj)}. It should be one of {self.allowed_contents}.") - if self._contains(obj): - raise ValueError("Cannot add this object because it is already contained within the list") - + def _set_parent(self, obj): # set the child-parent relationship if self.parent: relationship_name = self.parent.__class__.__name__.lower() @@ -49,11 +53,8 @@ def _handle_append(self, obj): setattr(obj, relationship_name, self.parent) def _contains(self, obj): - if self._items is None: - obj_ids = [] - else: - obj_ids = [id(item) for item in self._items] - return id(obj) in obj_ids + # membership is by identity + return self._items is not None and any(item is obj for item in self._items) def __str__(self): return str(self._items) @@ -75,12 +76,10 @@ def __radd__(self, objects): return objects + self._items def __contains__(self, key): - return key in self._items + return self._contains(key) def __iadd__(self, objects): - for obj in objects: - self._handle_append(obj) - self._items.extend(objects) + self.extend(objects) return self def __iter__(self): @@ -93,25 +92,55 @@ def __len__(self): return len(self._items) def __setitem__(self, key, value): - self._items[key] = value + if isinstance(key, slice): + values = list(value) + for obj in values: + self._check_allowed(obj) + # assign to a copy first, so that nothing changes if the assignment or the checks fail + new_items = self._items[:] + new_items[key] = values + if len({id(obj) for obj in new_items}) != len(new_items): + raise ValueError("Cannot add this object because it is already contained within the list") + for obj in values: + self._set_parent(obj) + self._items = new_items + else: + if value is self._items[key]: + return + self._handle_append(value) + self._items[key] = value def append(self, obj): self._handle_append(obj) self._items.append(obj) def extend(self, objects): + # objects may be an iterator that can only be consumed once (e.g., a generator), and we need to iterate twice + objects = list(objects) + # check everything before changing anything, so that a failure leaves the list unchanged. + # Building the set of ids once, rather than calling _contains() for each object, keeps this O(n + m) + seen = {id(item) for item in self._items or ()} + for obj in objects: + self._check_allowed(obj) + if id(obj) in seen: + raise ValueError("Cannot add this object because it is already contained within the list") + seen.add(id(obj)) for obj in objects: - self._handle_append(obj) + self._set_parent(obj) self._items.extend(objects) def clear(self): self._items = [] def count(self, value): - return self._items.count(value) + return int(self._contains(value)) def index(self, value, start=0, stop=sys.maxsize): - return self._items.index(value, start, stop) + items = self._items or [] + for i in range(*slice(start, stop).indices(len(items))): + if items[i] is value: + return i + raise ValueError("object is not in list") def insert(self, index, obj): self._handle_append(obj) @@ -121,10 +150,10 @@ def pop(self, index=-1): return self._items.pop(index) def remove(self, value): - return self._items.remove(value) + del self._items[self.index(value)] def reverse(self): - raise self._items.reverse() + self._items.reverse() def sort(self, *args, key=None, reverse=False): self._items.sort(*args, key=key, reverse=reverse) diff --git a/neo/core/spiketrainlist.py b/neo/core/spiketrainlist.py index 6b9859df3..887b028c4 100644 --- a/neo/core/spiketrainlist.py +++ b/neo/core/spiketrainlist.py @@ -105,10 +105,13 @@ def __init__(self, items=None, parent=None): if items is None: self._items = items else: + items = list(items) for item in items: if not is_spiketrain_or_proxy(item): raise ValueError("`items` can only contain SpikeTrain objects or proxy objects") - self._items = list(items) + if len({id(item) for item in items}) != len(items): + raise ValueError("`items` contains the same object more than once") + self._items = items self._spike_time_array = None self._channel_id_array = None self._all_channel_ids = None @@ -144,7 +147,7 @@ def __getitem__(self, i): def __setitem__(self, i, value): if self._items is None: self._spiketrains_from_array() - self._items[i] = value + super().__setitem__(i, value) def __str__(self): """Return str(self)""" @@ -183,7 +186,7 @@ def _add_spiketrainlists(self, other, in_place=False): if other._items is None: other._spiketrains_from_array() if in_place: - self._items.extend(other._items) + self.extend(other._items) return self else: return self.__class__(items=self._items[:] + other._items) @@ -232,11 +235,7 @@ def __iadd__(self, other): if isinstance(other, self.__class__): return self._add_spiketrainlists(other, in_place=True) elif other and is_spiketrain_or_proxy(other[0]): - for obj in other: - self._handle_append(obj) - if self._items is None: - self._spiketrains_from_array() - self._items.extend(other) + self.extend(other) return self else: raise TypeError("Can only add a SpikeTrainList or a list of SpikeTrains in place") @@ -246,11 +245,7 @@ def __radd__(self, other): if isinstance(other, self.__class__): return other._add_spiketrainlists(self) elif other and is_spiketrain_or_proxy(other[0]): - for obj in other: - obj.segment = self.segment - if self._items is None: - self._spiketrains_from_array() - self._items.extend(other) + self.extend(other) return self elif len(other) == 0: return self @@ -281,8 +276,7 @@ def append(self, obj): raise ValueError("Can only append SpikeTrain objects") if self._items is None: self._spiketrains_from_array() - self._handle_append(obj) - self._items.append(obj) + super().append(obj) def extend(self, iterable): """Extends the SpikeTrainList with additional SpikeTrain's from an iterable @@ -306,9 +300,7 @@ def extend(self, iterable): if self._items is None: self._spiketrains_from_array() - for obj in iterable: - self._handle_append(obj) - self._items.extend(iterable) + super().extend(iterable) @classmethod def from_spike_time_array( diff --git a/neo/test/coretest/test_container.py b/neo/test/coretest/test_container.py index 3ba8b2f5c..05a278709 100644 --- a/neo/test/coretest/test_container.py +++ b/neo/test/coretest/test_container.py @@ -268,6 +268,21 @@ def test_merge__unmergable_unequal_AssertionError(self): ) +class Test_Container_set_object_list(unittest.TestCase): + def test_set_list(self): + seg = neo.core.Segment() + sig = neo.core.AnalogSignal([1, 2, 3] * pq.mV, sampling_rate=1 * pq.kHz) + seg.analogsignals = [sig] + self.assertEqual(len(seg.analogsignals), 1) + self.assertIs(sig.segment, seg) + + def test_set_other_type_TypeError(self): + seg = neo.core.Segment() + sig = neo.core.AnalogSignal([1, 2, 3] * pq.mV, sampling_rate=1 * pq.kHz) + with self.assertRaises(TypeError): + seg.analogsignals = (sig,) + + @unittest.skipUnless(HAVE_IPYTHON, "requires IPython") class Test_pprint(unittest.TestCase): def test__pretty(self): diff --git a/neo/test/coretest/test_objectlist.py b/neo/test/coretest/test_objectlist.py new file mode 100644 index 000000000..847e3df4e --- /dev/null +++ b/neo/test/coretest/test_objectlist.py @@ -0,0 +1,162 @@ +""" +Tests of the neo.core.objectlist.ObjectList class +""" + +import unittest + +import quantities as pq + +from neo.core import AnalogSignal, Event, Group, Segment +from neo.core.objectlist import ObjectList + + +def make_signal(value=1.0): + return AnalogSignal([[value], [value + 1], [value + 2]] * pq.mV, sampling_rate=1 * pq.kHz) + + +class TestObjectList(unittest.TestCase): + def setUp(self): + self.segment = Segment() + self.objlist = ObjectList(AnalogSignal, parent=self.segment) + self.a, self.b, self.c = make_signal(1.0), make_signal(10.0), make_signal(100.0) + + def test_append_sets_parent(self): + self.objlist.append(self.a) + self.assertEqual(len(self.objlist), 1) + self.assertIs(self.a.segment, self.segment) + + def test_append_duplicate(self): + self.objlist.append(self.a) + self.assertRaises(ValueError, self.objlist.append, self.a) + self.assertEqual(len(self.objlist), 1) + + def test_append_equal_but_distinct_object(self): + # membership is by identity, so an object with the same data is a different object + self.objlist.append(self.a) + self.objlist.append(make_signal(1.0)) + self.assertEqual(len(self.objlist), 2) + + def test_append_wrong_type(self): + self.assertRaises(TypeError, self.objlist.append, Event()) + + def test_extend_from_generator(self): + self.objlist.extend(sig for sig in (self.a, self.b)) + self.assertEqual(len(self.objlist), 2) + self.assertIs(self.objlist[1], self.b) + + def test_iadd_from_generator(self): + self.objlist += (sig for sig in (self.a, self.b)) + self.assertEqual(len(self.objlist), 2) + + def test_extend_duplicate_within_iterable(self): + self.assertRaises(ValueError, self.objlist.extend, [self.a, self.a]) + self.assertEqual(len(self.objlist), 0) + self.assertIsNone(self.a.segment) + + def test_extend_duplicate_of_existing_item(self): + self.objlist.append(self.a) + self.assertRaises(ValueError, self.objlist.extend, [self.b, self.a]) + self.assertEqual(len(self.objlist), 1) + self.assertIsNone(self.b.segment) + + def test_extend_is_atomic(self): + self.assertRaises(TypeError, self.objlist.extend, [self.a, Event()]) + self.assertEqual(len(self.objlist), 0) + self.assertIsNone(self.a.segment) + + def test_contains(self): + self.objlist.extend([self.a, self.b]) + self.assertIn(self.a, self.objlist) + self.assertIn(self.b, self.objlist) + self.assertNotIn(self.c, self.objlist) + self.assertNotIn(make_signal(10.0), self.objlist) + + def test_index(self): + self.objlist.extend([self.a, self.b, self.c]) + self.assertEqual(self.objlist.index(self.c), 2) + self.assertEqual(self.objlist.index(self.c, 1), 2) + self.assertEqual(self.objlist.index(self.c, -1), 2) + self.assertRaises(ValueError, self.objlist.index, self.a, 1) + self.assertRaises(ValueError, self.objlist.index, make_signal(10.0)) + + def test_count(self): + self.objlist.extend([self.a, self.b]) + self.assertEqual(self.objlist.count(self.b), 1) + self.assertEqual(self.objlist.count(make_signal(10.0)), 0) + + def test_remove(self): + self.objlist.extend([self.a, self.b, self.c]) + self.objlist.remove(self.b) + self.assertEqual(list(self.objlist), [self.a, self.c]) + self.assertRaises(ValueError, self.objlist.remove, self.b) + + def test_append_after_removal(self): + self.objlist.extend([self.a, self.b]) + self.objlist.remove(self.a) + self.objlist.append(self.a) + self.assertIs(self.objlist.pop(), self.a) + self.objlist.append(self.a) + self.objlist.clear() + self.objlist.append(self.a) + self.assertEqual(len(self.objlist), 1) + + def test_setitem(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1] = self.c + self.assertIs(self.objlist[1], self.c) + self.assertIs(self.c.segment, self.segment) + + def test_setitem_same_object(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1] = self.b + self.assertEqual(list(self.objlist), [self.a, self.b]) + + def test_setitem_duplicate(self): + self.objlist.extend([self.a, self.b]) + with self.assertRaises(ValueError): + self.objlist[1] = self.a + self.assertEqual(list(self.objlist), [self.a, self.b]) + + def test_setitem_wrong_type(self): + self.objlist.append(self.a) + with self.assertRaises(TypeError): + self.objlist[0] = Event() + + def test_setitem_slice(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1:] = [self.b, self.c] + self.assertEqual(list(self.objlist), [self.a, self.b, self.c]) + self.assertIs(self.c.segment, self.segment) + + def test_setitem_slice_duplicate(self): + self.objlist.extend([self.a, self.b]) + with self.assertRaises(ValueError): + self.objlist[1:] = [self.c, self.a] + with self.assertRaises(ValueError): + self.objlist[1:] = [self.c, self.c] + self.assertEqual(list(self.objlist), [self.a, self.b]) + self.assertIsNone(self.c.segment) + + def test_setitem_slice_wrong_type(self): + self.objlist.append(self.a) + with self.assertRaises(TypeError): + self.objlist[:] = [self.b, Event()] + self.assertEqual(list(self.objlist), [self.a]) + + def test_reverse(self): + self.objlist.extend([self.a, self.b, self.c]) + self.objlist.reverse() + self.assertEqual(list(self.objlist), [self.c, self.b, self.a]) + + def test_without_parent(self): + group = Group() + objlist = ObjectList(AnalogSignal) + objlist.extend([self.a, self.b]) + self.assertEqual(len(objlist), 2) + self.assertIsNone(self.a.segment) + group.analogsignals.append(self.c) + self.assertIn(self.c, group.analogsignals) + + +if __name__ == "__main__": + unittest.main() diff --git a/neo/test/coretest/test_spiketrainlist.py b/neo/test/coretest/test_spiketrainlist.py index b1a9d517d..e3202a17c 100644 --- a/neo/test/coretest/test_spiketrainlist.py +++ b/neo/test/coretest/test_spiketrainlist.py @@ -268,6 +268,50 @@ def test_append_something_else(self): for stl in (self.stl_from_obj_list, self.stl_from_array): self.assertRaises(ValueError, stl.append, None) + def test_append_to_multiplexed(self): + """Appending to a SpikeTrainList created from an array should convert it to a list""" + new_st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.stl_from_array.append(new_st) + self.assertEqual(len(self.stl_from_array), 5) + self.assertIn(new_st, self.stl_from_array) + self.assertIs(self.stl_from_array[-1], new_st) + + def test_extend_from_generator(self): + """Extending from a generator should add all the SpikeTrains""" + for stl in (self.stl_from_obj_list, self.stl_from_array): + stl.extend( + SpikeTrain(times, units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) for times in ([22.2], [44.4]) + ) + self.assertEqual(len(stl), 6) + + def test_extend_duplicate(self): + """Extending with an iterable containing the same SpikeTrain twice should fail""" + new_st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.assertRaises(ValueError, self.stl_from_obj_list.extend, [new_st, new_st]) + self.assertEqual(len(self.stl_from_obj_list), 4) + self.assertRaises(ValueError, self.stl_from_obj_list.extend, [self.stl_from_obj_list[0]]) + self.assertEqual(len(self.stl_from_obj_list), 4) + + def test_create_with_duplicate(self): + """Creating a SpikeTrainList containing the same SpikeTrain twice should fail""" + st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.assertRaises(ValueError, SpikeTrainList, items=[st, st]) + + def test_create_from_generator(self): + stl = SpikeTrainList(items=(st for st in self.stl_from_obj_list)) + self.assertEqual(len(stl), 4) + + def test_add_duplicate(self): + """Combining SpikeTrainLists that share a SpikeTrain should fail, whichever operator is used""" + st = self.stl_from_obj_list[0] + self.assertRaises(ValueError, lambda: self.stl_from_obj_list + SpikeTrainList(items=[st])) + self.assertRaises(ValueError, lambda: self.stl_from_obj_list + [st]) + self.assertRaises(ValueError, lambda: [st] + self.stl_from_obj_list) + with self.assertRaises(ValueError): + self.stl_from_obj_list += SpikeTrainList(items=[st]) + self.assertEqual(len(self.stl_from_obj_list), 4) + self.assertEqual(self.stl_from_obj_list.count(st), 1) + def test_multiplexed(self): """The multiplexed property should return a pair of arrays""" channel_id_array, spike_time_array = self.stl_from_array.multiplexed