added test for voice separation

Author: CarlosCancino-Chacon

partitura/music_utils/__init__.py

@@ -0,0 +1 @@
+from .voice_separation import estimate_voices as voice_estimation

partitura/music_utils/voice_separation/__init__.py

@@ -0,0 +1 @@
+from .vosa import estimate_voices

…voice_separation/score_representation.py …usic_utils/voice_separation/_vs_utils.pypartitura/music_utils/voice_separation/score_representation.py renamed to partitura/music_utils/voice_separation/_vs_utils.py partitura/music_utils/voice_separation/score_representation.py renamed to partitura/music_utils/voice_separation/_vs_utils.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 """
-Base classes to represent score elements
+Base classes to represent score elements for voice separation
 
 TODO
 ----
@@ -12,11 +12,11 @@
 import numpy as np
 from collections import defaultdict
 
-from madmom.io.midi import MIDIFile
+# from madmom.io.midi import MIDIFile
 
 from statistics import mode
 
-SCORE_DTYPES = [('pitch', 'i4'), ('onset', 'f4'), ('duration', 'f4')]
+from ...utils import add_field
 
 
 def sort_by_pitch(sounding_notes):
@@ -106,10 +106,6 @@ def __init__(self, pitch, onset, duration, note_id, velocity=None, voice=None):
 
         self.is_grace = self.duration == 0
         self._grace = None
-
-        # I think this is no longer used... remove?
-        self.array = np.array([(self.pitch, self.onset, self.duration)],
-                              dtype=SCORE_DTYPES)
         self._voice = voice
         self.velocity = None
 
@@ -495,14 +491,26 @@ def has_voice_info(self):
 
 
 class VSScore(VSBaseScore):
-    """Class to represent a score
+    """Class to represent a score for voice separation
+
+    TODO:
+    * rename this class or simplify to avoid overlap in naming
+      conventions with the main package
+    * better handle grace notes
     """
 
     def __init__(self, score, delete_gracenotes=True):
 
+        # Score
         self.score = score
 
+        # Get the IDs of the notes
+        if 'id' not in self.score.dtype.names:
+            self.score = add_field(self.score, [('id', int)])
+            self.score['id'] = np.arange(len(self.score), dtype=int)
+
         if delete_gracenotes:
+            # TODO: Handle grace notes correctly
             self.score = self.score[score['duration'] != 0]
         else:
             grace_note_idxs = np.where(score['duration'] == 0)[0]
@@ -520,12 +528,12 @@ def __init__(self, score, delete_gracenotes=True):
 
         self.notes = []
 
-        for i, n in enumerate(self.score):
+        for n in self.score:
 
             note = VSNote(pitch=n['pitch'],
                           onset=n['onset'],
                           duration=n['duration'],
-                          note_id=i,
+                          note_id=n['id'],
                           velocity=n['velocity'] if 'velocity' in self.score.dtype.names else None)
 
             self.notes.append(note)
@@ -543,31 +551,31 @@ def __init__(self, score, delete_gracenotes=True):
 
         self.contigs = None
 
-    def write_midi(self, outfile, tempo=120, default_vel=60, skip_notes_wo_voice=True):
-        note_info = []
+    # def write_midi(self, outfile, tempo=120, default_vel=60, skip_notes_wo_voice=True):
+    #     note_info = []
 
-        if skip_notes_wo_voice:
-            out_notes = [n for n in self.notes if n.voice is not None]
-        else:
-            out_notes = self.notes
-        for n in out_notes:
+    #     if skip_notes_wo_voice:
+    #         out_notes = [n for n in self.notes if n.voice is not None]
+    #     else:
+    #         out_notes = self.notes
+    #     for n in out_notes:
 
-            pitch = n.pitch
-            onset = n.onset
-            duration = n.duration
-            velocity = n.velocity if n.velocity is not None else default_vel
-            channel = n.voice if n.voice is not None else 0
+    #         pitch = n.pitch
+    #         onset = n.onset
+    #         duration = n.duration
+    #         velocity = n.velocity if n.velocity is not None else default_vel
+    #         channel = n.voice if n.voice is not None else 0
 
-            note_info.append((onset, pitch, duration, velocity, channel))
+    #         note_info.append((onset, pitch, duration, velocity, channel))
 
-        note_info = np.array(sorted(note_info), dtype=np.float)
-        # onsets start at 0 (MIDI files cannot represent anacrusis)
-        note_info[:, 0] -= note_info[:, 0].min()
+    #     note_info = np.array(sorted(note_info), dtype=np.float)
+    #     # onsets start at 0 (MIDI files cannot represent anacrusis)
+    #     note_info[:, 0] -= note_info[:, 0].min()
 
-        mf = MIDIFile.from_notes(note_info,
-                                 tempo=tempo,
-                                 unit='beats')
-        mf.save(outfile)
+    #     mf = MIDIFile.from_notes(note_info,
+    #                              tempo=tempo,
+    #                              unit='beats')
+    #     mf.save(outfile)
 
     def write_txt(self, outfile, skip_notes_wo_voice=False):
 
@@ -588,13 +596,23 @@ def write_txt(self, outfile, skip_notes_wo_voice=False):
 
     @property
     def note_array(self):
-
+        """
+        TODO:
+        Check that all notes have the same type of id
+        """
         out_array = []
+
         for n in self.notes:
-            out_note = (n.pitch, n.onset, n.duration, n.voice if n.voice is not None else -1)
+            out_note = (n.pitch, n.onset, n.duration,
+                        n.voice if n.voice is not None else -1, n.id)
             out_array.append(out_note)
 
-        return np.array(out_array)
+        return np.array(out_array, dtype=[('pitch', 'i4'),
+                                          ('onset', 'f4'),
+                                          ('duration', 'f4'),
+                                          ('voice', 'i4'),
+                                          ('id', type(self.notes[0].id))]
+                        )
 
     def make_contigs(self):
 

partitura/music_utils/voice_separation/vosa.py

@@ -11,11 +11,8 @@
 from collections import defaultdict
 import numpy.ma as ma
 
-try:
-    from score_representation import VSScore, VSNote, VoiceManager, Contig, NoteStream
-except ModuleNotFoundError:
-    from .score_representation import VSScore, VSNote, VoiceManager, Contig, NoteStream
-
+from ._vs_utils import VSScore, VSNote, VoiceManager, Contig, NoteStream
+from ...utils import add_field
 
 # Maximal cost of a jump (in Chew  and Wu (2006) is 2 ** 31)
 MAX_COST = 1000
@@ -36,7 +33,7 @@ def pairwise_cost(prev, nxt):
     Returns
     -------
     cost : np.ndarray
-        Cost of connecting each note in the last onset of the 
+        Cost of connecting each note in the last onset of the
         previous of the contig to each note in the first onset
         of the next contig. (index [i, j] represents the cost
         from connecting i in `prev` to  j in `nxt`.
@@ -285,13 +282,67 @@ def estimate_voices(self):
                 keep_loop = False
 
 
-if __name__ == '__main__':
+def estimate_voices(notearray):
+    """
+    Voice estimation using the voice separation algorithm
+    proposed in [1]
+
+    Parameters
+    ----------
+    notearray : numpy structured array
+        Structured array containing score information.
+        Required fields are `pitch` (MIDI pitch),
+        `onset` (starting time of the notes) and
+        `duration` (duration of the notes). Additionally,
+        It might be useful to have an `id` field containing
+        the ID's of the notes. If this field is not contained
+        in the array, ID's will be created for the notes.
+
+    Returns
+    -------
+    v_notearray : numpy structured array
+        Structured array containing score information.
+        This array contains `pitch`, `onset`, `duration`
+        `voice` and `id`.
+
+    References
+    ----------
+    [1] Elaine Chew and Xiaodan Wu (2006) Separating Voices in
+        Polyphonic Music: A Contig Mapping Approach.
+
+    TODO
+    ----
+    * Handle grace notes correctly. The current version simply
+      deletes all grace notes.
+    """
+
+    if notearray.dtype.fields is None:
+        raise ValueError('`notearray` must be a structured numpy array')
+
+    for field in ('pitch', 'onset', 'duration'):
+        if field not in notearray.dtype.names:
+            raise ValueError('Input array does not contain the field {0}'.format(field))
+
+    if 'id' not in notearray.dtype.names:
+        print("The input score does not contain note ID's. They will be created")
+
+        input_array = add_field(notearray, [('id', int)])
+        input_array['id'] = np.arange(len(notearray), dtype=int)
+
+    else:
+        input_array = notearray
+
+    # Indices for sorting the output such that it matches
+    # the original input (VoSA reorders the notes by onset and
+    # by pitch)
+    orig_idxs = np.arange(len(notearray))
 
-    import partitura
+    id_sort_idxs = np.sort(input_array['id'])
 
-    fn = './Three-Part_Invention_No_13_(fragment).musicxml'
+    # Perform voice separation
+    v_notearray = VoSA(input_array[id_sort_idxs]).note_array
 
-    xml = partitura.musicxml.xml_to_notearray(fn)
+    # Sort output according to the note id's
+    v_notearray = v_notearray[v_notearray['id'].argsort()]
 
-    vsa = VoSA(xml)
-    vsa.write_midi('test.mid')
+    return v_notearray[np.argsort(orig_idxs[id_sort_idxs])]

partitura/utils.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python
 
+import numpy as np
 from functools import wraps
 from collections import defaultdict
 
+
 def cached_property(func, name=None):
     """
     cached_property(func, name=None) -> a descriptor
@@ -23,14 +25,14 @@ def cached_property(func, name=None):
     >>> # subsequent access of foo (fast):
     >>> x.foo
     """
-    if name is None :
+    if name is None:
         name = func.__name__
 
     @wraps(func)
     def _get(self):
-        try :
+        try:
             return self.__dict__[name]
-        except KeyError :
+        except KeyError:
             self.__dict__[name] = func(self)
             return self.__dict__[name]
 
@@ -39,13 +41,13 @@ def _set(self, value):
         self.__dict__[name] = value
 
     @wraps(func)
-    def _del(self ):
+    def _del(self):
         self.__dict__.pop(name, None)
 
     return property(_get, _set, _del)
 
 
-def iter_subclasses(cls, _seen = None):
+def iter_subclasses(cls, _seen=None):
     """
     iter_subclasses(cls)
 
@@ -69,14 +71,15 @@ def iter_subclasses(cls, _seen = None):
     >>> [cls.__name__ for cls in iter_subclasses(object)] #doctest: +ELLIPSIS
     ['type', ...'tuple', ...]
     """
-    
+
     if not isinstance(cls, type):
         raise TypeError('iter_subclasses must be called with '
                         'new-style classes, not %.100r' % cls)
-    if _seen is None: _seen = set()
+    if _seen is None:
+        _seen = set()
     try:
         subs = cls.__subclasses__()
-    except TypeError: # fails only when cls is type
+    except TypeError:  # fails only when cls is type
         subs = cls.__subclasses__(cls)
     for sub in subs:
         if sub not in _seen:
@@ -90,6 +93,7 @@ class ComparableMixin(object):
     """source:
     http://regebro.wordpress.com/2010/12/13/python-implementing-rich-comparison-the-correct-way/
     """
+
     def _compare(self, other, method):
         try:
             return method(self._cmpkey(), other._cmpkey())
@@ -99,22 +103,22 @@ def _compare(self, other, method):
             return NotImplemented
 
     def __lt__(self, other):
-        return self._compare(other, lambda s,o: s < o)
+        return self._compare(other, lambda s, o: s < o)
 
     def __le__(self, other):
-        return self._compare(other, lambda s,o: s <= o)
+        return self._compare(other, lambda s, o: s <= o)
 
     def __eq__(self, other):
-       return self._compare(other, lambda s,o: s == o)
+        return self._compare(other, lambda s, o: s == o)
 
     def __ge__(self, other):
-        return self._compare(other, lambda s,o: s >= o)
+        return self._compare(other, lambda s, o: s >= o)
 
     def __gt__(self, other):
-        return self._compare(other, lambda s,o: s > o)
+        return self._compare(other, lambda s, o: s > o)
 
     def __ne__(self, other):
-        return self._compare(other, lambda s,o: s != o)
+        return self._compare(other, lambda s, o: s != o)
 
 
 def partition(func, iterable):
@@ -125,5 +129,39 @@ def partition(func, iterable):
     """
     result = defaultdict(list)
     for v in iterable:
-        result[func(v)].append(v) 
+        result[func(v)].append(v)
     return result
+
+
+def add_field(a, descr):
+    """Return a new array that is like "a", but has additional fields.
+
+    From https://stackoverflow.com/questions/1201817/adding-a-field-to-a-structured-numpy-array
+
+    Arguments:
+      a     -- a structured numpy array
+      descr -- a numpy type description of the new fields
+
+    The contents of "a" are copied over to the appropriate fields in
+    the new array, whereas the new fields are uninitialized.  The
+    arguments are not modified.
+
+    >>> sa = np.array([(1, 'Foo'), (2, 'Bar')], \
+                         dtype=[('id', int), ('name', 'S3')])
+    >>> sa.dtype.descr == np.dtype([('id', int), ('name', 'S3')])
+    True
+    >>> sb = add_field(sa, [('score', float)])
+    >>> sb.dtype.descr == np.dtype([('id', int), ('name', 'S3'), \
+                                       ('score', float)])
+    True
+    >>> np.all(sa['id'] == sb['id'])
+    True
+    >>> np.all(sa['name'] == sb['name'])
+    True
+    """
+    if a.dtype.fields is None:
+        raise ValueError("`A` must be a structured numpy array")
+    b = np.empty(a.shape, dtype=a.dtype.descr + descr)
+    for name in a.dtype.names:
+        b[name] = a[name]
+    return b