Start the beat units again at each restart

packages/tilia-core/src/tilia_core/timemap/meters.py

@@ -6,8 +6,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
-from dataclasses import dataclass
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass, replace
 from fractions import Fraction
 
 from tilia_core import tla
@@ -29,7 +29,7 @@ class _BeatUnit:
     Its units are parsed when it is read, not again for each measure it governs.
     """
 
-    id: str
+    id: str | None  # None for a unit the core supplies at a movement's start
     denominator: int
     unit_values: tuple[Fraction, ...]
     assumed: bool
@@ -43,52 +43,79 @@ def meters_of(rows: Sequence[RowIn]) -> list[MeasureMeter]:
     makes it conflicting. `rows_of` has already left out the units that can't
     be read, so the unit before such a unit carries on.
 
-    Before the first unit, the measures take its denominator and units,
-    assumed; with no unit at all, a measure of N taps is N/4, assumed. Either
-    way measure 1 starts a unit, but none has a `beat_unit_id`, since no unit
-    sits on their beats. Only here does this differ from #621's loop, which
-    never meets such a timeline: #621's loader keeps a unit in the first
-    measure (`_ensure_first_beat_unit`), copied from the first unit or the
-    default.
+    Every movement starts with a unit: the timeline's first measure, and
+    each measure marked `restart`. Where a movement's first measure has none,
+    the core supplies one, assumed, with no `beat_unit_id`, since no unit
+    sits on those beats (`_supplied`). A movement's first measure always
+    starts a unit, and no unit reaches back across a restart. Only here does
+    this differ from #621's loop, which never meets such a timeline: #621's
+    loader keeps a unit in the first measure (`_ensure_first_beat_unit`),
+    copied from the first unit or the default, and #621 has no restarts.
     """
-    first = next((_parsed(*unit) for row in rows for unit in row.units), None)
-    if first is None:
-        opening = DEFAULT_DENOMINATOR, tuple(parse_units(DEFAULT_UNITS).units)
-    else:
-        opening = first.denominator, first.unit_values
-
     meters = []
     governing: _BeatUnit | None = None
-    for index, row in enumerate(rows):
-        beat_count = len(row.beat_ids)
-        in_measure = [_parsed(beat_id, unit) for beat_id, unit in row.units]
-        if in_measure:
-            governing = in_measure[0]
-
-        if governing is None:
-            denominator, units = opening[0], list(opening[1])
-        else:
-            denominator, units = governing.denominator, list(governing.unit_values)
-
-        meters.append(
-            MeasureMeter(
-                denominator=denominator,
-                units=tuple(units),
-                beat_values=tuple(fit_units(units, beat_count)),
-                beat_unit_id=governing.id if governing else None,
-                starts_here=bool(in_measure) or index == 0,
-                is_ambiguous=is_fit_ambiguous(units, beat_count),
-                is_conflicting=len(in_measure) > 1,
-                is_assumed=governing is None or governing.assumed,
+    for movement in _movements(rows):
+        units_set = [[_parsed(*unit) for unit in row.units] for row in movement]
+        first = next((unit for in_measure in units_set for unit in in_measure), None)
+        governing = _supplied(first or governing)
+        for index, (row, in_measure) in enumerate(
+            zip(movement, units_set, strict=True)
+        ):
+            if in_measure:
+                governing = in_measure[0]
+            beat_count = len(row.beat_ids)
+            units = list(governing.unit_values)
+            meters.append(
+                MeasureMeter(
+                    denominator=governing.denominator,
+                    units=tuple(units),
+                    beat_values=tuple(fit_units(units, beat_count)),
+                    beat_unit_id=governing.id,
+                    starts_here=bool(in_measure) or index == 0,
+                    is_ambiguous=is_fit_ambiguous(units, beat_count),
+                    is_conflicting=len(in_measure) > 1,
+                    is_assumed=governing.assumed,
+                )
             )
-        )
     return meters
 
 
+def _movements(rows: Sequence[RowIn]) -> Iterator[list[RowIn]]:
+    """The rows by movement: one starts at the first row and at each restart."""
+    movement: list[RowIn] = []
+    for row in rows:
+        if row.restart and movement:
+            yield movement
+            movement = []
+        movement.append(row)
+    if movement:
+        yield movement
+
+
+def _supplied(unit: _BeatUnit | None) -> _BeatUnit:
+    """
+    The unit the core supplies where a movement's first measure has none: a
+    copy of `unit` (the movement's first, else the one in force before it),
+    else the default. Assumed, since nobody set it there, and with no id,
+    since no beat carries it.
+    """
+    if unit is None:
+        return _DEFAULT
+    return replace(unit, id=None, assumed=True)
+
+
 def _parsed(beat_id: str, unit: tla.BeatUnit) -> _BeatUnit:
     return _BeatUnit(
         id=beat_id,
         denominator=unit.denominator,
         unit_values=tuple(parse_units(unit.units).units),
         assumed=unit.assumed,
     )
+
+
+_DEFAULT = _BeatUnit(
+    id=None,
+    denominator=DEFAULT_DENOMINATOR,
+    unit_values=tuple(parse_units(DEFAULT_UNITS).units),
+    assumed=True,
+)

packages/tilia-core/tests/timemap/test_timemap_meters.py

@@ -2,9 +2,10 @@
 
 import pytest
 
-from docs import beat_timeline
+from docs import beat_timeline, document
 from tilia_core.timemap._rows import rows_of
 from tilia_core.timemap.meters import meters_of
+from tilia_core.timemap.table import build_rows
 
 # What TiLiA puts on the first beat, and WP3's migration on an old file's.
 DEFAULT = {"denominator": 4, "units": "1", "assumed": True}
@@ -274,3 +275,123 @@ def test_scenario(
         assert meter.is_ambiguous == ambiguous
         assert all(isinstance(value, F) for value in meter.beat_values)
         assert isinstance(meter.numerator, F)
+
+
+def piece(*movements):
+    """
+    Beats a second apart, in `movements` of measures `(taps, unit or None)`,
+    each unit on its measure's downbeat. Every movement after the first
+    starts with a restart.
+    """
+    beats = []
+    for movement, measures in enumerate(movements):
+        for measure, (taps, beat_unit) in enumerate(measures):
+            mark = {"restart": True} if movement and measure == 0 else {}
+            for tap in range(taps):
+                downbeat = tap == 0
+                beats.append(
+                    {
+                        "time": float(len(beats)),
+                        "measure": mark if downbeat else None,
+                        "beat_unit": dict(beat_unit)
+                        if downbeat and beat_unit
+                        else None,
+                    }
+                )
+    return beat_timeline(beats)
+
+
+def quarters_before(result, index):
+    return sum(m.quarter_length(v) for m in result[:index] for v in m.beat_values)
+
+
+class TestRestarts:
+    # A restart is a start: each movement starts with a unit, as the
+    # timeline does, and no unit reaches back across a restart.
+
+    def test_a_pickup_without_a_unit_takes_its_movements_first_unit(self):
+        # Movement 1 in 3/4 tapped in three; movement 2 in 6/8 tapped in two,
+        # with a one-tap pickup and the unit set on its bar 1 only.
+        timeline = piece(
+            [(3, unit(4, "1")), (3, None)],
+            [(1, None), (2, unit(8, "3")), (2, None)],
+        )
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [
+            (3, 4),
+            (3, 4),
+            (3, 8),
+            (6, 8),
+            (6, 8),
+        ]
+        assert [m.is_assumed for m in result] == [False, False, True, False, False]
+        assert [m.starts_here for m in result] == [True, False, True, True, False]
+        assert result[2].beat_unit_id is None
+        assert result[3].beat_unit_id == beat_ids(timeline)[7]
+        assert quarters_before(result, 3) == F(15, 2)
+
+    def test_a_movement_without_a_unit_carries_the_one_before_assumed(self):
+        timeline = piece([(2, unit(8, "3")), (2, None)], [(2, None), (2, None)])
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [(6, 8)] * 4
+        assert [m.is_assumed for m in result] == [False, False, True, True]
+        assert [m.starts_here for m in result] == [True, False, True, False]
+        assert [m.beat_unit_id for m in result][2:] == [None, None]
+
+    def test_no_unit_reaches_back_across_a_restart(self):
+        timeline = piece([(2, None), (2, None)], [(2, unit(8, "3"))])
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [(2, 4), (2, 4), (6, 8)]
+        assert [m.is_assumed for m in result] == [True, True, False]
+
+    def test_a_restart_on_the_first_measure_changes_nothing(self):
+        timeline = tapped(8, (2,), {0: None, 6: unit(8, "3")})
+        without = meters(timeline)
+
+        timeline.measures.rows[0].restart = True
+
+        assert meters(timeline) == without
+
+    def test_a_movement_that_starts_with_its_own_unit(self):
+        timeline = piece([(2, unit(4, "1"))], [(2, unit(8, "3")), (2, None)])
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [(2, 4), (6, 8), (6, 8)]
+        assert not any(m.is_assumed for m in result)
+        assert result[1].beat_unit_id == beat_ids(timeline)[2]
+
+    def test_a_movement_without_a_unit_copies_a_supplied_one(self):
+        timeline = piece([(2, unit(8, "3"))], [(2, None)], [(2, None)])
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [(6, 8)] * 3
+        assert [m.is_assumed for m in result] == [False, True, True]
+        assert [m.starts_here for m in result] == [True, True, True]
+        assert [m.beat_unit_id for m in result][1:] == [None, None]
+
+    def test_the_unit_before_a_movement_is_the_last_one_in_force(self):
+        # Movement 2 starts with a supplied copy, then sets its own unit:
+        # movement 3, without one, carries movement 2's own unit, assumed.
+        timeline = piece(
+            [(4, unit(4, "1"))],
+            [(1, None), (2, unit(8, "3"))],
+            [(2, None)],
+        )
+
+        result = meters(timeline)
+        assert [m.time_signature for m in result] == [(4, 4), (3, 8), (6, 8), (6, 8)]
+        assert [m.is_assumed for m in result] == [False, True, False, True]
+
+    def test_the_table_reads_the_restarts(self):
+        timeline = piece(
+            [(3, unit(4, "1")), (3, None)],
+            [(1, None), (2, unit(8, "3")), (2, None)],
+        )
+
+        rows = build_rows(document(timeline), timeline)
+        assert [r.movement for r in rows] == [1, 1, 2, 2, 2]
+        assert [r.time_signature for r in rows] == ["3/4", "3/4", "3/8", "6/8", "6/8"]
+        assert [r.assumed for r in rows] == [False, False, True, False, False]