diff --git a/src/razorback/signalset.py b/src/razorback/signalset.py index 869c3f0..a3ecd51 100644 --- a/src/razorback/signalset.py +++ b/src/razorback/signalset.py @@ -2,10 +2,8 @@ """ -import warnings import itertools from collections import Counter -from collections.abc import MutableMapping from functools import reduce from datetime import datetime import fnmatch @@ -18,12 +16,6 @@ __all__ = ['SignalSet', 'SyncSignal', 'Tags', 'Inventory'] -try: - basestring -except NameError: - basestring = str - - def _tupleit(value): try: return tuple(value) @@ -31,85 +23,26 @@ def _tupleit(value): return (value,) -class TagsBase(MutableMapping): - " Base class for Tags " - def __init__(self, indices, dct): - self.__dict__.update(_tags={}, _indices=frozenset(indices)) - self.update(dct) - - @property - def indices(self): - return self._indices - - def __eq__(self, other): - return (self._tags == getattr(other, '_tags', None) - and self.indices == getattr(other, 'indices', None)) - - __req__ = __eq__ - - def __iter__(self): - return iter(self._tags) - - def __len__(self): - return len(self._tags) - - def __delitem__(self, key): - del self._tags[key] - - def __getitem__(self, key): - return self._tags[key] - - def __setitem__(self, key, value): - assert isinstance(key, basestring) - value = _tupleit(value) - assert value, "at least one index must be given" - if not all(v in self.indices for v in value): - msg = "values %s should be in indices %s" - raise ValueError(msg % (value, self.indices)) - self._tags[key] = value - - def __getattribute__(self, name): - try: - tags = super(TagsBase, self).__getattribute__('_tags') - except AttributeError: - tags = None - if tags and (name in tags): - return tags[name] - return super(TagsBase, self).__getattribute__(name) - - def __setattr__(self, name, value): - if name.startswith('_') or name in ['indices']: - return super(TagsBase, self).__setattr__(name, value) - self[name] = value - - def __delattr__(self, name): - try: - del self[name] - except KeyError: - super(TagsBase, self).__delattr__(name) - - -class Tags(TagsBase): +class Tags: """ - Tags(n, name1=idx1, name2=idx2, ...) - Tags(name1=idx1, name2=idx2, ...) + Tags(data) + Tags(data, size=n) Tags handle mappings between names and groups of indices. - A Tags object behaves like a dict but its keys must be legal string names - and its values are tuples of integers in a fixed range. - New entries of a Tags object are converted in tuples if needed - then their content is checked to belong in the fixed range. - - Item access can be emulated through attribute access: + `data`: { str: int | tuple[int] } + a dict whose keys are strings and the values are integers or tuples of integers. + keys are the tags and integers are the corresponding indices. - - tags.name1 --> tags['name1'] - - tags.name1 = idx1 --> tags['name1'] = idx1 + `size`: integer | None + the size of the index range. + if None, `size` is deduced from `data` values. + A Tags object behaves like a read-only dict with a few extra features. - Tags also implements the union (|) operator according to the SignalSet class: + Tags implements the union (|) operator against SyncSignal object to produce SignalSet objects. - - tags | signal -> SignalSet(tags, signal) + - tags | sync_signal -> SignalSet(tags, sync_signal) See Also @@ -117,29 +50,66 @@ class Tags(TagsBase): SignalSet """ - def __init__(self, n=None, **dct): - if n is None: - idx = sum(map(_tupleit, dct.values()), ()) - n = (1 + max(idx)) if idx else 0 - super(Tags, self).__init__(range(n), dct) + def __init__(self, data, *, size=None): + data = {k: _tupleit(v) for k, v in data.items()} + assert all(data.values()), "at least one index per tag must be provided" + assert all(isinstance(k, str) for k in data.keys()) + if size is None: + idx = sum(data.values(), ()) + size = (1 + max(idx)) if idx else 0 + else: + indices = range(size) + assert all(i in indices for v in data.values() for i in v), f"indices must be in range({size})" + assert size >= 0 + self._data = data + self._size = size + + @property + def size(self): + return self._size + + @property + def indices(self): + return set(range(self._size)) def __or__(self, signal): return SignalSet(self, signal) def __repr__(self): - items = sorted(self.items(), key=lambda x: (len(x[1]), x[0])) - tags = ', '.join('%s=%s' % e for e in items) - return '%s(%s, %s)' % (type(self).__name__, len(self.indices), tags) - - def __str__(self): - return str(self._tags) + return f"{type(self).__name__}({self._data}, size={self.size})" - # filter = fnmatch.filter def filter(self, *patterns): + """ use fnmatch.filter to select some tags + """ return tuple(set().union(*(fnmatch.filter(self, p) for p in patterns))) - def filter_get(self, *patterns): - return tuple(sorted(set().union(*(map(self.get, self.filter(*patterns)))))) + def __eq__(self, other): + return ( + isinstance(other, Tags) + and self._size == other._size + and self._data == other._data + ) + + def __getitem__(self, key): + return self._data[key] + + def __iter__(self): + return iter(self._data) + + def __len__(self): + return len(self._data) + + def get(self, key, default=None): + return self._data.get(key, default) + + def items(self): + return self._data.items() + + def keys(self): + return self._data.keys() + + def values(self): + return self._data.values() class SignalSet(object): @@ -408,7 +378,7 @@ def tags(self, value): if n and vidx is not None and not set(vidx).issubset(range(n)): msg = "incompatible tags" raise ValueError(msg) - self._tags = Tags(n or None, **value) + self._tags = Tags(value, size=n or None) def select_channels(self, channels): """ return a SignalSet containing the selected channels @@ -425,7 +395,7 @@ def select_channels(self, channels): channels = tuple(self.tags[channels]) elif isinstance(channels, slice): channels = tuple(range(self.nb_channels)[channels]) - elif isinstance(channels, basestring): + elif isinstance(channels, str): raise ValueError(msg % channels) else: channels = _tupleit(channels) @@ -444,7 +414,7 @@ def select_channels(self, channels): keep = set(indices).issuperset tags = {k: map(idxmap.get, _tupleit(v)) for k, v in self.tags.items() if keep(v)} - tags = Tags(len(indices), **tags) + tags = Tags(tags, size=len(indices)) signals = [s.select(*indices) for s in self.signals] return SignalSet(tags, *signals) @@ -498,7 +468,7 @@ def merge(self, *others): shift, N = shift[:-1], shift[-1] tags = {t: [i+c for i in ii] for c, s in zip(shift, seqs) for t, ii in s.tags.items()} - tags = Tags(N, **tags) + tags = Tags(tags, size=N) def rec(d, f, seq): "recursively find common intervals" diff --git a/tests/test_inventory.py b/tests/test_inventory.py index e881bee..e61cb3c 100644 --- a/tests/test_inventory.py +++ b/tests/test_inventory.py @@ -14,9 +14,9 @@ def test_simple(): sync = lambda t0, t1, rate=1, n=1: SyncSignal([np.arange(1+(t1-t0)*rate)]*n, rate, t0) inv = Inventory([ - Tags(4, Ex1=0, Ey1=1, Hx1=2, Hy1=3, E1=(0, 1), H1=(2, 3)) | sync(0, 20, n=4), - Tags(2, Hx2=0, Hy2=1, H2=(0, 1)) | sync(10, 20, n=2), - Tags(4, Ex2=0, Ey2=1, Hx2=2, Hy2=3, E2=(0, 1), H2=(2, 3)) | sync(20, 30, n=4), + Tags(dict(Ex1=0, Ey1=1, Hx1=2, Hy1=3, E1=(0, 1), H1=(2, 3)), size=4) | sync(0, 20, n=4), + Tags(dict(Hx2=0, Hy2=1, H2=(0, 1)), size=2) | sync(10, 20, n=2), + Tags(dict(Ex2=0, Ey2=1, Hx2=2, Hy2=3, E2=(0, 1), H2=(2, 3)), size=4) | sync(20, 30, n=4), ]) diff --git a/tests/test_signalset.py b/tests/test_signalset.py index 06dba3d..949409d 100644 --- a/tests/test_signalset.py +++ b/tests/test_signalset.py @@ -5,7 +5,7 @@ @pytest.fixture def tags(): - return Tags(a=0, b=1, c=2, BC=(1, 2)) + return Tags(dict(a=0, b=1, c=2, BC=(1, 2))) @pytest.fixture @@ -21,7 +21,7 @@ def three_signals(tags, data): sy3 = SyncSignal(10*data, 1.0, start=20) s1 = tags | sy1 - s2 = Tags(a=2, b=0, c=1, BC=(0, 1)) | sy2 + s2 = Tags(dict(a=2, b=0, c=1, BC=(0, 1))) | sy2 s3 = SignalSet(tags, sy3) return s1, s2, s3 @@ -30,7 +30,7 @@ def three_signals(tags, data): def test_base(tags, data, three_signals): s1, s2, s3 = three_signals - s = Tags(a=0, b=1, c=2) | SyncSignal(data, 1.0) + s = Tags(dict(a=0, b=1, c=2)) | SyncSignal(data, 1.0) s.tags.BC = 1, 2 a = s.select_channels([2, 0]) @@ -56,19 +56,19 @@ def test_indexing(tags, three_signals): sa = ss['a'] assert np.all(sa.intervals == ss.intervals) - assert sa.tags == Tags(1, a=0) + assert sa.tags == Tags(dict(a=0), size=1) sb = ss['b'] assert np.all(sb.intervals == ss.intervals) - assert sb.tags == Tags(1, b=0) + assert sb.tags == Tags(dict(b=0), size=1) sbc = ss[['b', 'c']] assert np.all(sbc.intervals == ss.intervals) - assert sbc.tags == Tags(2, b=0, c=1, BC=(0, 1)) + assert sbc.tags == Tags(dict(b=0, c=1, BC=(0, 1)), size=2) sb2 = ss[['b', 2]] assert np.all(sb2.intervals == ss.intervals) - assert sb2.tags == Tags(2, b=0, c=1, BC=(0, 1)) + assert sb2.tags == Tags(dict(b=0, c=1, BC=(0, 1)), size=2) assert ss[:].tags == ss.tags @@ -84,7 +84,7 @@ def test_indexing(tags, three_signals): channels = fnmatch.filter(ss.tags, '[bc]') sbc = ss[channels] assert np.all(sbc.intervals == ss.intervals) - assert sbc.tags == Tags(2, b=0, c=1, BC=(0, 1)) + assert sbc.tags == Tags(dict(b=0, c=1, BC=(0, 1)), size=2) def test_simple_fourier(): @@ -94,7 +94,7 @@ def test_simple_fourier(): x = np.cos(t*2*np.pi) y = np.sin(t*2*np.pi) - tags = Tags(x=0, y=1) + tags = Tags(dict(x=0, y=1)) sy = SyncSignal([x, y], 1/dt) sig = tags | sy sigc = tags | sy.extract_i(0, 15) | sy.extract_i(15-1, 30)