Refactor Tags as a simple class.

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,128 +16,100 @@
 __all__ = ['SignalSet', 'SyncSignal', 'Tags', 'Inventory']
 
 
-try:
-    basestring
-except NameError:
-    basestring = str
-
-
 def _tupleit(value):
     try:
         return tuple(value)
     except TypeError:
         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
     --------
     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"

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),
     ])
 
 

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)