Prevent consolidateCompletedTuplets() from breaking complete tuplets

music21/duration.py

@@ -3535,6 +3535,72 @@ def fixBrokenTupletDuration(self, tupletGroup: list[note.GeneralNote]) -> None:
                 n.duration.informClient()
         # else: pass
 
+
+class TupletSearchState:  # pylint: disable=W0201
+    '''
+    Private helper for makeNotation.consolidateCompletedTuplets().
+    '''
+    def __init__(self, onlyIfTied=True) -> None:
+        self.onlyIfTied: bool = onlyIfTied
+        self.reset()
+
+    def reset(self) -> None:
+        self.to_consolidate: list[note.GeneralNote | None] = []
+        self.partial_tuplet_sum: OffsetQL = 0.0
+        self.last_tuplet: Tuplet|None = None
+        self.completion_target: OffsetQL|None = None
+
+    def advance_tuplet_sum(self, gn: note.GeneralNote) -> None:
+        self.partial_tuplet_sum = opFrac(self.partial_tuplet_sum + gn.quarterLength)
+
+    def append(self, gn: note.GeneralNote) -> None:
+        if self.to_consolidate:
+            self.to_consolidate.append(gn)
+        else:
+            self.partial_tuplet_sum = gn.quarterLength
+            if not gn.duration.tuplets:
+                raise ValueError
+            self.last_tuplet = gn.duration.tuplets[0]
+            if t.TYPE_CHECKING:
+                assert self.last_tuplet is not None
+            self.completion_target = self.last_tuplet.totalTupletLength()
+            self.to_consolidate.append(gn)
+
+    def mark_no_consolidation(self) -> None:
+        self.to_consolidate.append(None)
+
+    def get_consolidatable_notes(self) -> list[note.GeneralNote]:
+        if not all(self.is_reexpressible(gn) for gn in self.to_consolidate):
+            return []
+        return t.cast(list['note.GeneralNote'], self.to_consolidate)
+
+    def is_reexpressible(self, gn: note.GeneralNote | None) -> bool:
+        return (
+            gn is not None
+            and gn.duration.expressionIsInferred
+            and len(gn.duration.tuplets) < 2
+            and (gn.isRest or gn.tie is not None or not self.onlyIfTied)
+        )
+
+    def should_be_tested(self, gn: note.GeneralNote) -> bool:
+        if not self.to_consolidate:
+            return True
+        prev_gn = gn.previous('GeneralNote', activeSiteOnly=True)
+        return (
+            (
+                # rests_match?
+                (gn.isRest and prev_gn.isRest)
+                # notes match?
+                or (not gn.isRest and not prev_gn.isRest and gn.pitches == prev_gn.pitches)
+            )
+            # And no gaps.
+            and opFrac(prev_gn.offset + prev_gn.quarterLength) == gn.offset
+            # And tuplet matches.
+            and len(gn.duration.tuplets) == 1
+            and gn.duration.tuplets[0] == self.last_tuplet
+        )
+
+
 # -------------------------------------------------------------------------------
 
 

music21/stream/makeNotation.py

@@ -2047,6 +2047,7 @@ def consolidateCompletedTuplets(
         - be consecutive (with respect to :class:`~music21.note.GeneralNote` objects)
         - be all rests, or all :class:`~music21.note.NotRest`s with equal `.pitches`
         - all have :attr:`~music21.duration.Duration.expressionIsInferred` = `True`.
+        - not begin during a tuplet
         - sum to the tuplet's total length
         - if `NotRest`, all must be tied (if `onlyIfTied` is True)
 
@@ -2065,7 +2066,7 @@ def consolidateCompletedTuplets(
     >>> [el.quarterLength for el in s.notesAndRests]
     [0.5, Fraction(1, 6), Fraction(1, 6), Fraction(1, 6)]
 
-    `mustBeTied` is `True` by default:
+    `onlyIfTied` is `True` by default:
 
     >>> s2 = stream.Stream()
     >>> n = note.Note(quarterLength=1/3)
@@ -2092,71 +2093,38 @@ def consolidateCompletedTuplets(
 
     Does nothing if there are multiple (nested) tuplets.
     '''
-    def is_reexpressible(gn: note.GeneralNote) -> bool:
-        return (
-            gn.duration.expressionIsInferred
-            and len(gn.duration.tuplets) < 2
-            and (gn.isRest or gn.tie is not None or not onlyIfTied)
-        )
-
+    search_state = duration.TupletSearchState(onlyIfTied=onlyIfTied)
     iterator: Iterable[stream.Stream]
     if recurse:
         iterator = s.recurse(streamsOnly=True, includeSelf=True)
     else:
         iterator = [s]
     for container in iterator:
-        reexpressible = [gn for gn in container.notesAndRests if is_reexpressible(gn)]
-        to_consolidate: list[note.GeneralNote] = []
-        partial_tuplet_sum: OffsetQL = 0.0
-        last_tuplet: duration.Tuplet|None = None
-        completion_target: OffsetQL|None = None
-        for gn in reexpressible:
-            prev_gn = gn.previous(note.GeneralNote, activeSiteOnly=True)
-            if (
-                prev_gn in to_consolidate
-                and (
-                    (isinstance(gn, note.Rest) and isinstance(prev_gn, note.Rest))
-                    or (
-                        isinstance(gn, note.NotRest)
-                        and isinstance(prev_gn, note.NotRest)
-                        and gn.pitches == prev_gn.pitches
-                    )
-                )
-                and opFrac(prev_gn.offset + prev_gn.quarterLength) == gn.offset
-                and len(gn.duration.tuplets) == 1 and gn.duration.tuplets[0] == last_tuplet
-            ):
-                partial_tuplet_sum = opFrac(partial_tuplet_sum + gn.quarterLength)
-                to_consolidate.append(gn)
-
-                if partial_tuplet_sum == completion_target:
+        search_state.reset()
+        for gn in container.notesAndRests:
+            search_state.advance_tuplet_sum(gn)
+            if search_state.should_be_tested(gn):
+                try:
+                    search_state.append(gn)
+                except ValueError:
+                    # Not in a tuplet, keep scanning.
+                    pass
+            elif search_state.to_consolidate:
+                # Found during an incomplete tuplet, but doesn't match it.
+                search_state.mark_no_consolidation()
+
+            if search_state.partial_tuplet_sum == search_state.completion_target:
+                if consolidatableNotes := search_state.get_consolidatable_notes():
                     # set flag to remake tuplet brackets
                     container.streamStatus.tuplets = False
-                    first_note_in_group = to_consolidate[0]
-                    for other_note in to_consolidate[1:]:
+                    first_note_in_group = consolidatableNotes[0]
+                    for other_note in consolidatableNotes[1:]:
                         container.remove(other_note)
                     first_note_in_group.duration.clear()
                     first_note_in_group.duration.tuplets = ()
-                    first_note_in_group.quarterLength = completion_target
+                    first_note_in_group.quarterLength = search_state.completion_target
+                search_state.reset()
 
-                    # reset search values
-                    to_consolidate = []
-                    partial_tuplet_sum = 0.0
-                    last_tuplet = None
-                    completion_target = None
-            else:
-                # reset to current values
-                if gn.duration.tuplets:
-                    partial_tuplet_sum = gn.quarterLength
-                    last_tuplet = gn.duration.tuplets[0]
-                    if t.TYPE_CHECKING:
-                        assert last_tuplet is not None
-                    completion_target = last_tuplet.totalTupletLength()
-                    to_consolidate = [gn]
-                else:
-                    to_consolidate = []
-                    partial_tuplet_sum = 0.0
-                    last_tuplet = None
-                    completion_target = None
 
 @contextlib.contextmanager
 def saveAccidentalDisplayStatus(s) -> t.Generator[None, None, None]:
@@ -2374,6 +2342,21 @@ def testMakeTiesChangingTimeSignatures(self):
         self.assertEqual(len(pp[stream.Measure][2].notes), 1)
         self.assertEqual(pp[stream.Measure][2].notes.first().duration.quarterLength, 24.0)
 
+    def testConsolidateCompletedTupletsNoFalsePositive(self):
+        from fractions import Fraction
+        from music21 import converter
+
+        s = converter.parse('tinyNotation: 2/4 trip{c8 d8 e8} trip{e8 e8 r8}')
+        for el in s[note.GeneralNote]:
+            el.duration.expressionIsInferred = True
+        consolidateCompletedTuplets(s, recurse=True, onlyIfTied=False)
+
+        # Before, the 3 e8's were consolidated, breaking both tuplets.
+        self.assertEqual(
+            [gn.quarterLength for gn in s[note.GeneralNote]],
+            [Fraction(1, 3)] * 6,
+        )
+
     def testSaveAccidentalDisplayStatus(self):
         from music21 import interval
         from music21 import stream