From 0e71e225ab4f82ed8ba620e7d096eccdc163c6c6 Mon Sep 17 00:00:00 2001 From: azfoo <45888544+azfoo@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:38:22 +0200 Subject: [PATCH 1/2] feat(core): start the beat units again at each restart A restart, the first measure of a new movement or song in the same file, now starts the beat units as the timeline's first measure does. Where a movement's first measure has no unit, meters_of supplies one, assumed and with no beat_unit_id: a copy of the movement's first unit, else of the unit in force before the movement, else the default. A movement's first measure always starts a unit, and no unit reaches back across a restart, so a first movement without a unit stays N/4 though a later one has a unit. A restart on the first measure changes nothing, and a timeline without restarts reads as before. Before, a later movement whose pickup had no unit took the previous movement's unit, not assumed, and every quarter stamp after it was off. Part of TimeLineAnnotator/project-management#323 Co-Authored-By: Claude Opus 5.5 --- .../src/tilia_core/timemap/meters.py | 67 +++++++++----- .../tests/timemap/test_timemap_meters.py | 88 +++++++++++++++++++ 2 files changed, 133 insertions(+), 22 deletions(-) diff --git a/packages/tilia-core/src/tilia_core/timemap/meters.py b/packages/tilia-core/src/tilia_core/timemap/meters.py index bbc890970..5d446ac99 100644 --- a/packages/tilia-core/src/tilia_core/timemap/meters.py +++ b/packages/tilia-core/src/tilia_core/timemap/meters.py @@ -6,7 +6,7 @@ from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Iterator, Sequence from dataclasses import dataclass from fractions import Fraction @@ -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,48 +43,71 @@ 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 (`_opening`). 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] + starts_movement = index == 0 or row.restart if in_measure: governing = in_measure[0] + elif starts_movement or governing is None: + governing = _opening(rows, index, governing) - if governing is None: - denominator, units = opening[0], list(opening[1]) - else: - denominator, units = governing.denominator, list(governing.unit_values) + 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, + beat_unit_id=governing.id, + starts_here=bool(in_measure) or starts_movement, is_ambiguous=is_fit_ambiguous(units, beat_count), is_conflicting=len(in_measure) > 1, - is_assumed=governing is None or governing.assumed, + is_assumed=governing.assumed, ) ) return meters +def _opening(rows: Sequence[RowIn], start: int, before: _BeatUnit | None) -> _BeatUnit: + """ + The unit the core supplies at the start of the movement that begins at + `start`, whose first measure has none: a copy of the movement's first + unit, else of the unit in force `before` it, else the default. Assumed, + since nobody set it there, and with no id, since no beat carries it. + """ + first = next((unit for row in _movement(rows, start) for unit in row.units), None) + if first is not None: + source = _parsed(*first) + denominator, values = source.denominator, source.unit_values + elif before is not None: + denominator, values = before.denominator, before.unit_values + else: + denominator = DEFAULT_DENOMINATOR + values = tuple(parse_units(DEFAULT_UNITS).units) + return _BeatUnit(id=None, denominator=denominator, unit_values=values, assumed=True) + + +def _movement(rows: Sequence[RowIn], start: int) -> Iterator[RowIn]: + """The rows from `start` up to the next restart.""" + yield rows[start] + for row in rows[start + 1 :]: + if row.restart: + return + yield row + + def _parsed(beat_id: str, unit: tla.BeatUnit) -> _BeatUnit: return _BeatUnit( id=beat_id, diff --git a/packages/tilia-core/tests/timemap/test_timemap_meters.py b/packages/tilia-core/tests/timemap/test_timemap_meters.py index 9e9bd4560..c967949c3 100644 --- a/packages/tilia-core/tests/timemap/test_timemap_meters.py +++ b/packages/tilia-core/tests/timemap/test_timemap_meters.py @@ -274,3 +274,91 @@ 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.components[beat_ids(timeline)[0]].attrs["measure"] = {"restart": True} + 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] From a22d15d18bed45a8538b092456a7b58bad69a839 Mon Sep 17 00:00:00 2001 From: azfoo <45888544+azfoo@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:56:19 +0200 Subject: [PATCH 2/2] refactor(core): compute meters movement by movement meters_of now splits the rows into movements once and runs the rule for the start of a timeline on each, with the unit in force before as the fallback. Before, each movement start without a unit scanned and copied the rest of the table, which made meters_of quadratic in the number of movements (20000 one-measure movements: 0.46 s, now 0.06 s); it also parsed a movement's first unit twice and kept a condition that could never be true. The tests add a third movement that copies a supplied unit, a movement that carries the unit set in the one before it, and the measure table's rows for a restart; the restart on the first measure is now set on the row only. Part of TimeLineAnnotator/project-management#323 Co-Authored-By: Claude Opus 5.5 --- .../src/tilia_core/timemap/meters.py | 100 +++++++++--------- .../tests/timemap/test_timemap_meters.py | 37 ++++++- 2 files changed, 87 insertions(+), 50 deletions(-) diff --git a/packages/tilia-core/src/tilia_core/timemap/meters.py b/packages/tilia-core/src/tilia_core/timemap/meters.py index 5d446ac99..4c2ce907b 100644 --- a/packages/tilia-core/src/tilia_core/timemap/meters.py +++ b/packages/tilia-core/src/tilia_core/timemap/meters.py @@ -7,7 +7,7 @@ from __future__ import annotations from collections.abc import Iterator, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from fractions import Fraction from tilia_core import tla @@ -46,7 +46,7 @@ def meters_of(rows: Sequence[RowIn]) -> list[MeasureMeter]: 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 (`_opening`). A movement's first measure always + 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`), @@ -54,58 +54,54 @@ def meters_of(rows: Sequence[RowIn]) -> list[MeasureMeter]: """ 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] - starts_movement = index == 0 or row.restart - if in_measure: - governing = in_measure[0] - elif starts_movement or governing is None: - governing = _opening(rows, index, governing) - - 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, - starts_here=bool(in_measure) or starts_movement, - is_ambiguous=is_fit_ambiguous(units, beat_count), - is_conflicting=len(in_measure) > 1, - is_assumed=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 _opening(rows: Sequence[RowIn], start: int, before: _BeatUnit | None) -> _BeatUnit: +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 at the start of the movement that begins at - `start`, whose first measure has none: a copy of the movement's first - unit, else of the unit in force `before` it, else the default. Assumed, - since nobody set it there, and with no id, since no beat carries it. + 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. """ - first = next((unit for row in _movement(rows, start) for unit in row.units), None) - if first is not None: - source = _parsed(*first) - denominator, values = source.denominator, source.unit_values - elif before is not None: - denominator, values = before.denominator, before.unit_values - else: - denominator = DEFAULT_DENOMINATOR - values = tuple(parse_units(DEFAULT_UNITS).units) - return _BeatUnit(id=None, denominator=denominator, unit_values=values, assumed=True) - - -def _movement(rows: Sequence[RowIn], start: int) -> Iterator[RowIn]: - """The rows from `start` up to the next restart.""" - yield rows[start] - for row in rows[start + 1 :]: - if row.restart: - return - yield row + if unit is None: + return _DEFAULT + return replace(unit, id=None, assumed=True) def _parsed(beat_id: str, unit: tla.BeatUnit) -> _BeatUnit: @@ -115,3 +111,11 @@ def _parsed(beat_id: str, unit: tla.BeatUnit) -> _BeatUnit: 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, +) diff --git a/packages/tilia-core/tests/timemap/test_timemap_meters.py b/packages/tilia-core/tests/timemap/test_timemap_meters.py index c967949c3..206a5fd60 100644 --- a/packages/tilia-core/tests/timemap/test_timemap_meters.py +++ b/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} @@ -350,7 +351,6 @@ 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.components[beat_ids(timeline)[0]].attrs["measure"] = {"restart": True} timeline.measures.rows[0].restart = True assert meters(timeline) == without @@ -362,3 +362,36 @@ def test_a_movement_that_starts_with_its_own_unit(self): 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]