product-vamp/generator/performance.py

"""Page timing and supplied recordings. No note synthesis or audio transformation."""
from __future__ import annotations

import fractions
import hashlib
import json
import math
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)
  measures = document['staves'][0]['measures']
  systems = []
  for system in page.systems:
    a, b = times.index(system.start), times.index(system.end)
    labels = {p.sources[0]: p for p in system.primitives if p.role == 'measure_number'}
    boxes = []
    for i in range(a, b):
      box = labels[measures[i]['id']].box
      boxes.append(dict(measure=i + 1, bounds=[box.left, box.top, box.right, box.bottom]))
    systems.append(dict(first_measure=a + 1, measure_count=b - a, start_beat=generator.model.rational(system.start), end_beat=generator.model.rational(system.end), labels=boxes))
  return dict(first_measure=first + 1, measure_count=last - first, start_beat=generator.model.rational(start), end_beat=generator.model.rational(end), systems=systems)

def validate_views(document, pages, playback_pages):
  validate_pages(document, pages)
  times = boundaries(document)
  canonical = []

  def system_range(system):
    a, count = system['first_measure'], system['measure_count']
    if type(a) is not int or type(count) is not int or a < 1 or count < 1 or a + count > len(times):
      raise ValueError('invalid system measure coverage')
    if generator.model.fraction(system['start_beat']) != times[a - 1] or generator.model.fraction(system['end_beat']) != times[a + count - 1]:
      raise ValueError('invalid system timing')
    if [label['measure'] for label in system['labels']] != list(range(a, a + count)):
      raise ValueError('invalid measure label coverage')
    for label in system['labels']:
      box = label['bounds']
      if not isinstance(box, list) or len(box) != 4 or any(type(v) not in (int, float) or not math.isfinite(v) for v in box) or not (4 <= box[0] < box[2] <= 972 and 4 <= box[1] < box[3] <= 480):
        raise ValueError('invalid measure label bounds')
    return a, count

  for page in pages:
    systems = page.get('systems', [])
    if len(systems) != 2 and (page is not pages[-1] or len(systems) != 1):
      raise ValueError('invalid page systems')
    cursor = page['first_measure']
    for system in systems:
      a, count = system_range(system)
      if a != cursor:
        raise ValueError('system gap or overlap')
      cursor += count
      canonical.append(system)
    if cursor != page['first_measure'] + page['measure_count']:
      raise ValueError('systems disagree with page coverage')
  if not isinstance(playback_pages, list) or len(playback_pages) != len(canonical):
    raise ValueError('missing playback pages; regenerate score assets')
  for i, page in enumerate(playback_pages):
    expected = canonical[i:i + 2]
    systems = page.get('systems', [])
    if len(systems) != len(expected):
      raise ValueError('invalid playback systems')
    for actual, source in zip(systems, expected):
      system_range(actual)
      if any(actual[k] != source[k] for k in ('first_measure', 'measure_count', 'start_beat', 'end_beat')):
        raise ValueError('playback changes system boundaries')
    if any(page[k] != expected[0][k] for k in ('first_measure', 'start_beat')) or page['end_beat'] != expected[-1]['end_beat'] or page['measure_count'] != sum(s['measure_count'] for s in expected):
      raise ValueError('invalid playback page timing')

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']
  ending = [i for i, m in enumerate(measures) if m.get('ending_sustain')]
  if ending and ending != list(range(ending[0], len(measures))):
    raise ValueError('held ending must be a contiguous terminal section')
  sustain_start = times[ending[0]] if ending else times[-1]
  # The written held ending participates in page time, not metronome time.
  clicks = []
  for start, measure in zip(times, measures):
    if start >= sustain_start:
      break
    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]),
    sustain_start=generator.model.rational(sustain_start), 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")