Playback

generator/performance.py

@@ -0,0 +1,110 @@
+"""Page timing and supplied recordings. No note synthesis or audio transformation."""
+from __future__ import annotations
+
+import fractions
+import hashlib
+import json
+import pathlib
+import re
+import shutil
+import wave
+
+import generator.model
+
+TEMPOS = tuple(range(60, 121, 10))
+
+def boundaries(document):
+  result = [fractions.Fraction()]
+  for measure in document['staves'][0]['measures']:
+    result.append(result[-1] + generator.model.fraction(measure['duration']))
+  return result
+
+def page_timing(document, page):
+  times = boundaries(document)
+  start, end = page.systems[0].start, page.systems[-1].end
+  first, last = times.index(start), times.index(end)
+  return dict(first_measure=first + 1, measure_count=last - first, start_beat=generator.model.rational(start), end_beat=generator.model.rational(end))
+
+def validate_pages(document, pages):
+  times = boundaries(document)
+  cursor = 0
+  for page in pages:
+    if not isinstance(page, dict):
+      raise ValueError('invalid page descriptor')
+    count = page.get('measure_count')
+    if type(count) is not int or count <= 0 or page.get('first_measure') != cursor + 1 or cursor + count >= len(times):
+      raise ValueError('invalid page measure coverage')
+    if generator.model.fraction(page.get('start_beat')) != times[cursor] or generator.model.fraction(page.get('end_beat')) != times[cursor + count]:
+      raise ValueError('page timing does not match its measures')
+    cursor += count
+  if cursor != len(times) - 1:
+    raise ValueError('incomplete page measure coverage')
+
+def timing(document):
+  times = boundaries(document)
+  measures = document['staves'][0]['measures']
+  # Metronome accents follow each actual measure, including meter changes.
+  clicks = []
+  for start, measure in zip(times, measures):
+    duration = generator.model.fraction(measure['duration'])
+    beat = fractions.Fraction()
+    while beat < duration:
+      clicks.append(dict(beat=generator.model.rational(start + beat), downbeat=beat == 0))
+      beat += 1
+  return dict(song_id=document['metadata'].get('song_id', ''),
+    duration=generator.model.rational(times[-1]),
+    count_in_beats=4, clicks=clicks)
+
+def describe_wav(path, root, *, song_id='', bpm=0):
+  if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(root.resolve()):
+    raise ValueError(f'{path}: expected a contained regular WAV')
+  try:
+    with wave.open(str(path), 'rb') as audio:
+      channels, rate, frames = audio.getnchannels(), audio.getframerate(), audio.getnframes()
+      if audio.getcomptype() != 'NONE' or audio.getsampwidth() != 2 or channels not in (1, 2) or not 8000 <= rate <= 192000 or frames <= 0:
+        raise ValueError('expected nonempty mono/stereo PCM16 WAV')
+      remaining = frames
+      while remaining:
+        count = min(remaining, 65536)
+        if len(audio.readframes(count)) != count * channels * 2:
+          raise ValueError('truncated WAV data')
+        remaining -= count
+  except (wave.Error, EOFError) as error:
+    raise ValueError(f'{path}: invalid WAV: {error}') from error
+  with path.open('rb') as source:
+    digest = hashlib.file_digest(source, 'sha256').hexdigest()
+  return dict(path=path.relative_to(root).as_posix(), sha256=digest, channels=channels, sample_rate=rate, frames=frames, song_id=song_id, bpm=bpm)
+
+def copy_audio(static, destination, song_ids):
+  entries = []
+  inputs = [(static / 'sounds' / name, '', 0) for name in ('beat1.wav', 'beat2.wav')]
+  for path in sorted((static / 'backing').glob('*.wav')):
+    match = re.fullmatch(r'([a-z0-9]+(?:_[a-z0-9]+)*)_(60|70|80|90|100|110|120)\.wav', path.name)
+    if not match:
+      raise ValueError(f'{path}: unsupported backing filename or tempo')
+    if match[1] not in song_ids:
+      continue
+    inputs.append((path, match[1], int(match[2])))
+  for source, song_id, bpm in inputs:
+    description = describe_wav(source, static, song_id=song_id, bpm=bpm)
+    target = destination / description['path']
+    target.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copyfile(source, target)
+    entries.append(description)
+  return entries
+
+def validate_recording_lengths(root, entries):
+  durations = {}
+  for path in sorted((root / 'songs').glob('*.json')):
+    document = json.loads(path.read_text())
+    performance = timing(document)
+    song_id = performance['song_id']
+    duration = generator.model.fraction(performance['duration']) + performance['count_in_beats']
+    if song_id in durations and durations[song_id] != duration:
+      raise ValueError(f'{song_id}: parts disagree about performance duration')
+    durations[song_id] = duration
+  for entry in entries:
+    if entry['song_id']:
+      seconds = fractions.Fraction(entry['frames'], entry['sample_rate'])
+      if seconds < durations[entry['song_id']] * 60 / entry['bpm']:
+        raise ValueError(f"{entry['path']}: backing recording is shorter than count-in and score")