Cleanupf

generator/__main__.py

@@ -10,7 +10,7 @@ from generator.arrange.generate import generate_performance_scores
 from generator.practice.generate import generate_practice_scores
 from generator.score import count_json_files, validate_generated_tree
 from generator.summary import validate_expected_performance_summary, validate_practice_parts
-from generator.musicxml import load_musicxml
+from generator.parts import source_paths
 from generator.engraving import engrave_tree
 from generator.playback import compile_tree
 from generator.catalogue import build_catalogue
@@ -50,11 +50,12 @@ def generate_assets(args: argparse.Namespace) -> None:
     if progress:
       print(f'Assets [{monotonic() - started:.0f}s]: {message}', flush=True)
 
-  report('validating MusicXML sources')
+  report('checking source paths')
   for source in sorted(data_dir.rglob('*.xml')):
     if source.parent not in (data_dir / 'practice', data_dir / 'songs'):
       raise ValueError(f'{source}: MusicXML sources belong directly under data/practice or data/songs')
-    load_musicxml(source)
+  source_paths(data_dir / 'songs')
+  source_paths(data_dir / 'practice', practice=True)
   output_dir.parent.mkdir(parents=True, exist_ok=True)
   with tempfile.TemporaryDirectory(prefix='.vamp-assets-', dir=output_dir.parent) as temporary:
     staged = Path(temporary) / 'assets'

generator/arrange/generate.py

@@ -6,19 +6,19 @@ from pathlib import Path
 
 from generator.arrange.form import full_arrangement_sections
 from generator.guitar import place_guitar
-from generator.parts import PART_TYPES, build_measure, propagated_chords, validate_source
+from generator.parts import PART_TYPES, build_measure, prepare_document, propagated_chords, source_paths
 from generator.model import fraction, key_label, make_event, meter_label, rational
 from generator.musicxml import load_musicxml
-from generator.pitch import concert_document
 from generator.score import copy_source_voices, make_arranged_document, reidentify, slugify, source_staff, write_json
 
 FINAL_SECTION_INDEX = 6
 
 def arrange_document(document: dict, part_type: str) -> dict:
+  return _arrange_document(prepare_document(document), part_type)
+
+def _arrange_document(document: dict, part_type: str) -> dict:
   if part_type not in PART_TYPES:
     raise ValueError(f'unknown arrangement part: {part_type}')
-  validate_source(document)
-  document = concert_document(document, octave_transposing=True)
   staff = source_staff(document)
   source = staff['measures']
   sections = full_arrangement_sections(source, part_type)
@@ -90,14 +90,15 @@ def arrange_document(document: dict, part_type: str) -> dict:
   return result
 
 def generate_performance_scores(source_dir: Path, output_dir: Path) -> int:
-  if not source_dir.exists():
-    raise ValueError(f'performance source directory not found: {source_dir}')
   count = 0
-  for path in sorted(source_dir.glob('*.xml')):
-    document = load_musicxml(path)
+  for path in source_paths(source_dir):
+    try:
+      document = prepare_document(load_musicxml(path))
+    except ValueError as error:
+      raise ValueError(f'{path}: {error}') from error
     for part in PART_TYPES:
       try:
-        write_json(output_dir / f'{slugify(path.stem)}-{part}.json', arrange_document(document, part))
+        write_json(output_dir / f'{slugify(path.stem)}-{part}.json', _arrange_document(document, part))
       except ValueError as error:
         raise ValueError(f'{path}, part {part}: {error}') from error
       count += 1

generator/engraving/layout.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 from dataclasses import replace
+from bisect import bisect_left, bisect_right
 from fractions import Fraction
 import math
 
 from generator import harmony
 from generator.engraving import tablature
-from generator.model import fraction, validate_document
+from generator.model import validate_document
 from generator.engraving.attributes import barline, staff_attributes, tempo
 from generator.engraving.fonts import Fonts
 from generator.engraving.notation import ScoreIndex
@@ -19,7 +20,7 @@ def lyric_text(verse: dict) -> str:
 
 def column(index: ScoreIndex, time: Fraction, start: Fraction, end: Fraction, fonts: Fonts, style: Style) -> tuple[list[Primitive], list[Anchor], list[Fragment]]:
   items, anchors, fragments = [], [], []
-  events = [e for e in index.events if e.start == time or (time == start and e.start < start < e.end)]
+  events = index.events_at(time, include_sustained=time == start)
   grace_width = 0
   occupied = []
   for event in events:
@@ -47,48 +48,45 @@ def column(index: ScoreIndex, time: Fraction, start: Fraction, end: Fraction, fo
         text = lyric_text(verse)
         if text:
           items.append(fonts.text(text, x + style.space / 2, lane * 23, 18, 'lyric', (event.value['id'] + ':lyric:' + verse['number'], event.value['id']), True))
-  for measure_start, measure_end, measure in index.measures:
-    if not measure_start <= time < measure_end:
-      continue
-    for item in measure['harmonies']:
-      if measure_start + fraction(item['onset']) == time:
-        label = harmony.symbol(item)
-        items.append(fonts.text(label, 0, 0, 21, 'harmony', (item['id'],)))
-    for i, value in enumerate(measure.get('tempos', [])):
-      if measure_start + fraction(value['onset']) == time:
-        items.extend(tempo(value, fonts, 0, 0, style.space, measure['id'] + f':tempo:{i}'))
-    if time == measure_start:
-      items.append(fonts.text(str(measure['number']), 0, -style.space * (index.state(time)['staff_details']['lines'] + 0.5), 12, 'measure-number', (measure['id'],)))
-      if label := measure.get('section_label'):
-        items.append(fonts.text(label, 0, 0, 19, 'section', (measure['id'] + ':section',)))
+  measure_start, _, measure = index.measure_at(time)
+  for owner, _, item in index.harmonies_by_onset.get(time, []):
+    if owner is measure:
+      items.append(fonts.text(harmony.symbol(item), 0, 0, 21, 'harmony', (item['id'],)))
+  for owner, i, value in index.tempos_by_onset.get(time, []):
+    if owner is measure:
+      items.extend(tempo(value, fonts, 0, 0, style.space, measure['id'] + f':tempo:{i}'))
+  if time == measure_start:
+    items.append(fonts.text(str(measure['number']), 0, -style.space * (index.state(time)['staff_details']['lines'] + 0.5), 12, 'measure-number', (measure['id'],)))
+    if label := measure.get('section_label'):
+      items.append(fonts.text(label, 0, 0, 19, 'section', (measure['id'] + ':section',)))
   return items, anchors, fragments
 
 def make_system(index: ScoreIndex, start: Fraction, end: Fraction, fonts: Fonts, style: Style, justify: bool = False) -> System:
   items, anchors, fragments = [], [], []
   state = index.state(start)
-  measure = next(m for a, b, m in index.measures if a <= start < b)
+  measure = index.measure_at(start)[2]
   header = staff_attributes(index, state, start, fonts, style.space, measure['id'] + ':initial', ('clef', 'key', 'meter', 'staff_details'))
   items.extend(header)
   cursor = max((p.box.right for p in header), default=0) + 22
   content_left = cursor - 9
-  times = sorted({start} | {t for t in index.columns if start < t < end})
+  times = [start] + index.column_times[bisect_right(index.column_times, start):bisect_left(index.column_times, end)]
   placements = []
   groups = []
   for ti, time in enumerate(times):
     item_start, anchor_start = len(items), len(anchors)
     drawn, placed, pieces = column(index, time, start, end, fonts, style)
     leading = []
-    for measure_start, measure_end, m in index.measures:
-      if time == measure_start and time > start:
-        previous = next(previous for a, b, previous in index.measures if b == time)
-        leading.extend(barline(previous.get('barlines', {}).get('right', 'regular'), 0, -style.space * (index.state(time)['staff_details']['lines'] - 1), previous['id']))
-        old_state = index.state(time, before=True)
-        fields = tuple(k for k in ('clef', 'key', 'meter', 'staff_details') if old_state[k] != index.state(time)[k])
-        leading.extend(p.moved(12, 0) for p in staff_attributes(index, index.state(time), time, fonts, style.space, m['id'] + ':initial', fields))
-      for ci, change in enumerate(m.get('attributes', [])):
-        if measure_start + fraction(change['onset']) == time and time > start:
-          extra = staff_attributes(index, index.state(time), time, fonts, style.space, m['id'] + f':attributes:{ci}', tuple(change))
-          leading.extend(p.moved(max((p.box.right for p in leading), default=0) + 10, 0) for p in extra)
+    measure_start, _, m = index.measure_at(time)
+    if time == measure_start and time > start:
+      previous = index.measure_at(time, before=True)[2]
+      leading.extend(barline(previous.get('barlines', {}).get('right', 'regular'), 0, -style.space * (index.state(time)['staff_details']['lines'] - 1), previous['id']))
+      old_state = index.state(time, before=True)
+      fields = tuple(k for k in ('clef', 'key', 'meter', 'staff_details') if old_state[k] != index.state(time)[k])
+      leading.extend(p.moved(12, 0) for p in staff_attributes(index, index.state(time), time, fonts, style.space, m['id'] + ':initial', fields))
+    if time > start:
+      for owner, ci, change in index.attributes_by_onset.get(time, []):
+        extra = staff_attributes(index, index.state(time), time, fonts, style.space, owner['id'] + f':attributes:{ci}', tuple(change))
+        leading.extend(p.moved(max((p.box.right for p in leading), default=0) + 10, 0) for p in extra)
     if leading:
       width = bounds(leading).right - bounds(leading).left + 14
       items.extend(p.moved(cursor - bounds(leading).left, 0) for p in leading)
@@ -113,7 +111,7 @@ def make_system(index: ScoreIndex, start: Fraction, end: Fraction, fonts: Fonts,
     placements = [(time, x + i * stretch) for i, (time, x) in enumerate(placements)]
     width = available
   right = width - 5
-  last_measure = next(m for a, b, m in index.measures if a < end <= b)
+  last_measure = index.measure_at(end, before=True)[2]
   tab_items = [p for p in items if p.role.startswith('tab')]
   items = [p for p in items if not p.role.startswith('tab')]
   items = draw_beams(index, anchors, items, start, end, content_left, right, style)
@@ -200,7 +198,7 @@ def layout_score(document: dict, style: Style, fonts: Fonts) -> LayoutDocument:
   start = Fraction()
   available = style.width - style.margin_x * 2
   while start < index.end:
-    candidates = sorted(b for b in index.boundaries if b > start)
+    candidates = index.boundary_times[bisect_right(index.boundary_times, start):]
     selected = None
     # Reserve the same title/credit and footer bands when considering a split,
     # so a wide pitch range can break vertically as well as horizontally.
@@ -212,7 +210,7 @@ def layout_score(document: dict, style: Style, fonts: Fonts) -> LayoutDocument:
       selected = trial
     if selected is None:
       measure_end = candidates[0]
-      for end in sorted(t for t in index.columns if start < t < measure_end):
+      for end in index.column_times[bisect_right(index.column_times, start):bisect_left(index.column_times, measure_end)]:
         trial = make_system(index, start, end, fonts, style)
         if trial.box.right - trial.box.left <= available and trial.box.bottom - trial.box.top <= height_limit:
           selected = trial

generator/engraving/notation.py

@@ -26,6 +26,10 @@ class ScoreIndex:
     self.staff = document['staves'][0]
     self.measures = []
     self.events = []
+    self.events_by_onset = {}
+    self.attributes_by_onset = {}
+    self.harmonies_by_onset = {}
+    self.tempos_by_onset = {}
     self.by_id = {}
     self.tones = {}
     self.boundaries = {Fraction()}
@@ -48,18 +52,46 @@ class ScoreIndex:
           for tone in note['pitches']:
             self.tones[tone['id']] = (event, tone)
       for field in ('attributes', 'harmonies', 'tempos'):
-        self.columns.update(time + fraction(item['onset']) for item in measure.get(field, []))
+        bucket = {'attributes': self.attributes_by_onset, 'harmonies': self.harmonies_by_onset, 'tempos': self.tempos_by_onset}[field]
+        for index, item in enumerate(measure.get(field, [])):
+          onset = time + fraction(item['onset'])
+          self.columns.add(onset)
+          bucket.setdefault(onset, []).append((measure, index, item))
       time = end
     self.end = time
     self.measure_starts = [item[0] for item in self.measures]
     self.events.sort(key=lambda e: (e.start, not bool(e.value.get('grace')), e.lane))
+    self.event_starts = []
+    self.event_max_ends = []
+    for event in self.events:
+      self.events_by_onset.setdefault(event.start, []).append(event)
+      self.event_starts.append(event.start)
+      self.event_max_ends.append(max(event.end, self.event_max_ends[-1] if self.event_max_ends else event.end))
+    self.onsets = list(self.events_by_onset)
+    self.column_times = sorted(self.columns)
+    self.boundary_times = sorted(self.boundaries)
     self.lyric_lanes = list(dict.fromkeys((e.voice, verse['number']) for e in self.events for verse in e.value.get('lyrics', [])))
     self.accidentals = self.accidental_state()
+  def measure_at(self, time: Fraction, before: bool = False) -> tuple:
+    position = bisect_left(self.measure_starts, time) if before else bisect_right(self.measure_starts, time)
+    return self.measures[max(0, position - 1)]
+  def events_at(self, time: Fraction, include_sustained: bool = False) -> list[Event]:
+    attacks = self.events_by_onset.get(time, [])
+    if not include_sustained:
+      return attacks
+    sustained = []
+    position = bisect_left(self.event_starts, time) - 1
+    # Prefix maxima let system continuations stop searching once every earlier note has ended.
+    while position >= 0 and self.event_max_ends[position] > time:
+      event = self.events[position]
+      if event.end > time:
+        sustained.append(event)
+      position -= 1
+    return list(reversed(sustained)) + attacks
   def state(self, time: Fraction, before: bool = False) -> dict:
     if (time, before) in self._states:
       return self._states[time, before]
-    position = bisect_left(self.measure_starts, time) if before else bisect_right(self.measure_starts, time)
-    start, _, measure = self.measures[max(0, position - 1)]
+    start, _, measure = self.measure_at(time, before)
     state = {key: measure[key] for key in ('key', 'meter', 'clef', 'transpose', 'staff_details')}
     for change in measure.get('attributes', []):
       onset = start + fraction(change['onset'])
@@ -82,13 +114,13 @@ class ScoreIndex:
     for start, end, measure in self.measures:
       memory = {}
       key = None
-      onsets = sorted({e.start for e in self.events if start <= e.start < end})
+      onsets = self.onsets[bisect_left(self.onsets, start):bisect_left(self.onsets, end)]
       for onset in onsets:
         state = self.state(onset)
         if key != state['key']:
           key, memory = state['key'], {}
         simultaneous = {}
-        for event in (e for e in self.events if e.start == onset):
+        for event in self.events_by_onset[onset]:
           for tone in event.value['pitches']:
             hidden_head = isinstance(tone.get('notehead'), dict) and tone['notehead'].get('name') == 'none'
             if event.value.get('hidden') or tone.get('print', {}).get('print-object') == 'no' or hidden_head:
@@ -108,7 +140,7 @@ class ScoreIndex:
         for identity, values in simultaneous.items():
           memory[identity] = next(iter(values)) if len(values) == 1 else None
         if any(len(values) > 1 for values in simultaneous.values()):
-          for event in (e for e in self.events if e.start == onset):
+          for event in self.events_by_onset[onset]:
             for tone in event.value['pitches']:
               if 'letter' in tone and len(simultaneous.get((tone['letter'], tone['octave']), ())) > 1:
                 result[tone['id']] = True

generator/parts.py

@@ -1,14 +1,36 @@
 from __future__ import annotations
 
 from fractions import Fraction
+from pathlib import Path
 
 from generator.guitar import validate_tuning
 from generator.model import beat_lengths, fraction, make_event
-from generator.pitch import shift_pitch
-from generator.score import copy_source_voices, make_measure, measure_notes, source_staff
+from generator.pitch import concert_document, shift_pitch
+from generator.score import copy_source_voices, make_measure, slugify, source_staff
 
 PART_TYPES = ('acoustic', 'electric', 'vocal')
 
+def source_paths(directory: Path, *, practice: bool = False) -> list[Path]:
+  if not directory.is_dir():
+    raise ValueError(f'{"practice" if practice else "performance"} source directory not found: {directory}')
+  paths = sorted(directory.rglob('*.xml'))
+  identities = {}
+  for path in paths:
+    if practice:
+      if path.parent != directory or len(path.stem) != 3 or not path.stem.isascii() or not path.stem.isdecimal():
+        raise ValueError(f'{path}: practice sources require a three-digit filename directly under data/practice')
+    elif path.parent != directory:
+      raise ValueError(f'{path}: performance sources belong directly under data/songs')
+    identity = path.stem if practice else slugify(path.stem)
+    if identity in identities:
+      raise ValueError(f'{identities[identity]} and {path}: conflicting output path {identity}-acoustic.json')
+    identities[identity] = path
+  return paths
+
+def prepare_document(document: dict, *, practice: bool = False) -> dict:
+  validate_source(document, practice=practice)
+  return concert_document(document, octave_transposing=True)
+
 def propagated_chords(measures: list[dict], *, include_optional: bool = True) -> list[dict | None]:
   result = []
   current = None
@@ -34,8 +56,9 @@ def generated_voice(measure: dict, role: str, previous: dict | None) -> list[dic
   common_time = measure['time_signature'] == '4/4' and not any('meter' in a for a in measure.get('attributes', []))
   if role == 'rhythm' and common_time:
     attacks = [(Fraction(n, 2), Fraction(1, 2)) for n in (0, 3, 4, 5, 7)]
-  elif role == 'accompaniment' and common_time:
-    attacks = [(Fraction(1, 2), Fraction(1, 2))]
+  elif role == 'accompaniment':
+    boundaries = sorted({Fraction(), length} | {fraction(h['onset']) for h in measure['harmonies'] if fraction(h['onset']) < length})
+    attacks = [(start, end - start) for start, end in zip(boundaries, boundaries[1:])]
   else:
     attacks = []
     states = [(Fraction(), measure['meter'])]
@@ -59,11 +82,10 @@ def generated_voice(measure: dict, role: str, previous: dict | None) -> list[dic
       events.append(make_event(f'rest-{len(events)}', cursor, onset - cursor, 'rest'))
     duration = min(duration, length - onset)
     chord = active_chord(measure, onset, previous)
-    source_attack = any(fraction(n['onset']) == onset and n['kind'] == 'note' and not n.get('grace') and any(not p.get('tie_from') for p in n['pitches']) for n in measure_notes(measure))
     if role == 'slash':
       event = make_event(f'slash-{len(events)}', onset, duration, 'note')
       event['notehead'] = 'slash'
-    elif not chord or role == 'accompaniment' and source_attack:
+    elif not chord:
       event = make_event(f'rest-{len(events)}', onset, duration, 'rest')
     else:
       event = make_event(f'chord-{len(events)}', onset, duration, 'note')

generator/playback.py

@@ -75,11 +75,15 @@ def compile_timeline(document: dict) -> dict:
             raise PlaybackUnavailable('grace group must steal between zero and 100 percent')
           lengths = [fraction(donor['duration']) * amount / 100 for amount in percentages]
           total = sum(lengths, Fraction())
+          delay, extension = adjustments.get(donor['id'], (Fraction(), Fraction()))
           if mode == 'steal-time-previous':
             cursor -= total
-            adjustments[donor['id']] = (Fraction(), -total)
           else:
-            adjustments[donor['id']] = (total, -total)
+            delay += total
+          extension -= total
+          if fraction(donor['duration']) + extension <= 0 or any(fraction(p['duration']) + extension <= 0 for p in donor['pitches']):
+            raise PlaybackUnavailable(f'{donor["id"]}: grace timing consumes a donor or chord tone')
+          adjustments[donor['id']] = (delay, extension)
         for item, length in zip(pending, lengths):
           value, _, pitch_state, lane_id = item
           adjustments[value['id']] = (cursor - item[1], length)

generator/practice/generate.py

@@ -4,16 +4,16 @@ from pathlib import Path
 
 from generator.guitar import place_guitar
 from generator.musicxml import load_musicxml
-from generator.parts import PART_TYPES, build_measure, propagated_chords, validate_source
-from generator.pitch import concert_document
+from generator.parts import PART_TYPES, build_measure, prepare_document, propagated_chords, source_paths
 from generator.scat import add_scat
 from generator.score import make_arranged_document, reidentify, source_staff, write_json
 
 def practice_document(document: dict, part_type: str) -> dict:
+  return _practice_document(prepare_document(document, practice=True), part_type)
+
+def _practice_document(document: dict, part_type: str) -> dict:
   if part_type not in PART_TYPES:
     raise ValueError(f'unknown practice part: {part_type}')
-  validate_source(document, practice=True)
-  document = concert_document(document, octave_transposing=True)
   staff = source_staff(document)
   previous = propagated_chords(staff['measures'])
   role = {'acoustic': 'accompaniment', 'electric': 'melody', 'vocal': 'vocal'}[part_type]
@@ -30,17 +30,15 @@ def practice_document(document: dict, part_type: str) -> dict:
   return result
 
 def generate_practice_scores(source_dir: Path, output_dir: Path) -> int:
-  if not source_dir.is_dir():
-    raise ValueError(f'practice source directory not found: {source_dir}')
-  sources = sorted(source_dir.rglob('*.xml'))
+  sources = source_paths(source_dir, practice=True)
   for path in sources:
-    if path.parent != source_dir or len(path.stem) != 3 or not path.stem.isascii() or not path.stem.isdecimal():
-      raise ValueError(f'{path}: practice sources require a three-digit filename directly under data/practice')
-  for path in sources:
-    document = load_musicxml(path)
+    try:
+      document = prepare_document(load_musicxml(path), practice=True)
+    except ValueError as error:
+      raise ValueError(f'{path}: {error}') from error
     for part in PART_TYPES:
       try:
-        write_json(output_dir / path.stem / (part + '.json'), practice_document(document, part))
+        write_json(output_dir / path.stem / (part + '.json'), _practice_document(document, part))
       except ValueError as error:
         raise ValueError(f'{path}, part {part}: {error}') from error
   return len(sources) * len(PART_TYPES)

generator/score.py

@@ -114,7 +114,7 @@ def reidentify(measures: list[dict], source_tuplets: list[dict]) -> list[dict]:
   return tuplets
 
 def make_arranged_document(source_document: dict, staff: dict, *, staff_id: str, staff_name: str, role: str, arrangement_part_type: str, measures: list[dict], tuplets=None, octave_shifts=None) -> dict:
-  return {'schema_version': SCHEMA_VERSION, 'metadata': copy.deepcopy(source_document['metadata']), 'staves': [{'id': staff_id, 'name': staff_name, 'clef': copy.deepcopy(staff['clef']), 'role': role, 'arranged': True, 'arrangement_part_type': arrangement_part_type, 'measures': measures, 'tuplets': tuplets or [], 'octave_shifts': octave_shifts or [], 'instruments': copy.deepcopy(staff.get('instruments', []))}]}
+  return {'schema_version': SCHEMA_VERSION, 'metadata': copy.deepcopy(source_document['metadata']), 'staves': [{'id': staff_id, 'name': staff_name, 'clef': copy.deepcopy(staff['clef']), 'role': role, 'arranged': True, 'arrangement_part_type': arrangement_part_type, 'measures': measures, 'tuplets': tuplets or [], 'octave_shifts': copy.deepcopy(octave_shifts or []), 'instruments': copy.deepcopy(staff.get('instruments', []))}]}
 
 def count_json_files(root: Path) -> int:
   return sum(1 for p in root.rglob('*.json') if p.is_file())