khodowany
264 lines
13 KiB
Python
264 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from array import array
|
|
from fractions import Fraction
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
import wave
|
|
|
|
from generator.model import TYPE_LENGTHS, fraction, pitch_number, rational
|
|
from generator.repeats import expand_document
|
|
|
|
SAMPLE_RATE = 48000
|
|
PROFILE = 'soft-struck-1'
|
|
MAX_EVENTS = 1000000
|
|
|
|
class PlaybackUnavailable(ValueError):
|
|
pass
|
|
|
|
def sounding_pitch(pitch: dict, state: dict, guitar: bool) -> Fraction:
|
|
if 'unpitched' in pitch:
|
|
raise PlaybackUnavailable('unpitched notation has no generated-tone interpretation')
|
|
if guitar:
|
|
import generator.practice.guitar
|
|
value = generator.practice.guitar.sounding_pitch(pitch, state)
|
|
else:
|
|
transpose = state['transpose']
|
|
value = pitch_number(pitch) + transpose['chromatic'] + 12 * transpose['octave_change']
|
|
frequency = 440 * 2 ** (float(value - 69) / 12)
|
|
if not 20 <= frequency <= 18000:
|
|
raise PlaybackUnavailable(f'sounding pitch {value} is outside the generated-tone range (20-18000 Hz)')
|
|
return value
|
|
|
|
def compile_timeline(document: dict, *, guitar: bool = False) -> dict:
|
|
document = expand_document(document)
|
|
events, notes, linked = [], {}, {}
|
|
absolute = Fraction()
|
|
for measure in document['staves'][0]['measures']:
|
|
for voice in measure['voices']:
|
|
state = {key: measure[key] for key in ('transpose', 'clef', 'staff_details')}
|
|
changes = iter(sorted(measure.get('attributes', []), key=lambda a: fraction(a['onset'])))
|
|
change = next(changes, None)
|
|
for note in voice['notes']:
|
|
onset = fraction(note['onset'])
|
|
while change is not None and fraction(change['onset']) <= onset:
|
|
state.update({key: change[key] for key in state if key in change})
|
|
change = next(changes, None)
|
|
notes[note['id']] = (note, absolute + onset, dict(state), voice['id'])
|
|
absolute += fraction(measure['duration'])
|
|
adjustments = {}
|
|
insertions = {}
|
|
lanes = {}
|
|
for item in notes.values():
|
|
lanes.setdefault(item[3], []).append(item)
|
|
for lane in lanes.values():
|
|
pending = []
|
|
previous = None
|
|
for note, onset, state, voice in lane:
|
|
if note.get('grace'):
|
|
pending.append((note, onset, state, voice))
|
|
continue
|
|
if pending:
|
|
cursor = onset
|
|
specified = [n['grace'] for n, _, _, _ in pending]
|
|
modes = {key for g in specified for key in ('make-time', 'steal-time-previous', 'steal-time-following') if key in g}
|
|
if len(modes) > 1:
|
|
raise PlaybackUnavailable('grace group mixes conflicting timing modes')
|
|
mode = next(iter(modes), 'steal-time-following')
|
|
if mode == 'make-time':
|
|
lengths = [fraction(g.get(mode, rational(0))) for g in specified]
|
|
total = sum(lengths, Fraction())
|
|
insertions[onset] = max(insertions.get(onset, Fraction()), total)
|
|
else:
|
|
donor = previous[0] if mode == 'steal-time-previous' and previous else note if mode == 'steal-time-following' else None
|
|
if donor is None:
|
|
raise PlaybackUnavailable('grace notes have no preceding duration to steal')
|
|
percentages = [fraction(g[mode]) if mode in g else Fraction(25, len(pending)) for g in specified]
|
|
if not 0 < sum(percentages) < 100:
|
|
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())
|
|
if mode == 'steal-time-previous':
|
|
cursor -= total
|
|
adjustments[donor['id']] = (Fraction(), -total)
|
|
else:
|
|
adjustments[donor['id']] = (total, -total)
|
|
for item, length in zip(pending, lengths):
|
|
value, _, pitch_state, lane_id = item
|
|
adjustments[value['id']] = (cursor - item[1], length)
|
|
cursor += length
|
|
pending = []
|
|
previous = (note, onset)
|
|
if pending:
|
|
raise PlaybackUnavailable('trailing grace group has no following anchor')
|
|
for identifier, (note, onset, state, voice) in notes.items():
|
|
if note.get('cue'):
|
|
continue
|
|
delay, extension = adjustments.get(identifier, (Fraction(), Fraction()))
|
|
performed = onset + delay + sum((length for time, length in insertions.items() if time < onset or time == onset and not note.get('grace')), Fraction())
|
|
tremolo = note.get('tremolo')
|
|
if tremolo and tremolo['type'] == 'stop':
|
|
continue
|
|
for pitch in note['pitches']:
|
|
if pitch.get('cue'):
|
|
continue
|
|
length = fraction(pitch['duration']) + extension
|
|
if not note.get('grace'):
|
|
length += sum((amount for time, amount in insertions.items() if onset < time < onset + fraction(pitch['duration'])), Fraction())
|
|
if length <= 0:
|
|
raise PlaybackUnavailable(f'{identifier}: grace timing consumes a chord tone')
|
|
value = sounding_pitch(pitch, state, guitar)
|
|
if pitch.get('tie_from'):
|
|
previous = linked.get(pitch['tie_from'])
|
|
if previous is None or fraction(previous['onset']) + fraction(previous['duration']) != performed or fraction(previous['pitch']) != value:
|
|
raise PlaybackUnavailable(f'{identifier}: tie has inconsistent performed timing or pitch')
|
|
previous['duration'] = rational(fraction(previous['duration']) + length)
|
|
linked[pitch['id']] = previous
|
|
continue
|
|
if tremolo and tremolo['marks']:
|
|
if pitch.get('tie_to'):
|
|
raise PlaybackUnavailable('tied measured tremolo is not yet qualified for playback')
|
|
written = pitch['written']
|
|
step = min(Fraction(1), TYPE_LENGTHS[written['type']]) * fraction(written.get('ratio', rational(1))) / 2 ** tremolo['marks']
|
|
if tremolo['type'] == 'start':
|
|
partner = notes[tremolo['partner']]
|
|
if partner[1] != onset + fraction(note['duration']) or len(partner[0]['pitches']) != len(note['pitches']):
|
|
raise PlaybackUnavailable('paired tremolo requires contiguous equal-size chords')
|
|
step *= 2
|
|
length += fraction(partner[0]['pitches'][note['pitches'].index(pitch)]['duration'])
|
|
cursor, attack = Fraction(), 0
|
|
while cursor < length:
|
|
amount = value
|
|
if tremolo['type'] == 'start' and attack % 2:
|
|
other = partner[0]['pitches'][note['pitches'].index(pitch)]
|
|
amount = sounding_pitch(other, partner[2], guitar)
|
|
events.append({'id': f'{pitch["id"]}-{attack}', 'pitch': rational(amount), 'onset': rational(performed + cursor), 'duration': rational(min(step, length - cursor))})
|
|
cursor += step
|
|
attack += 1
|
|
if len(events) > MAX_EVENTS:
|
|
raise PlaybackUnavailable('playback exceeds one million tone events')
|
|
else:
|
|
event = {'id': pitch['id'], 'pitch': rational(value), 'onset': rational(performed), 'duration': rational(length)}
|
|
events.append(event)
|
|
linked[pitch['id']] = event
|
|
if len(events) > MAX_EVENTS:
|
|
raise PlaybackUnavailable('playback exceeds one million tone events')
|
|
events.sort(key=lambda e: (fraction(e['onset']), e['id']))
|
|
return {'version': 1, 'available': True, 'duration': rational(absolute + sum(insertions.values(), Fraction())), 'events': events}
|
|
|
|
def write_pcm(path: Path, values) -> str:
|
|
samples = array('h', (round(max(-1, min(1, sample)) * 32767) for sample in values))
|
|
if sys.byteorder != 'little':
|
|
samples.byteswap()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with wave.open(str(path), 'wb') as output:
|
|
output.setparams((1, 2, SAMPLE_RATE, 0, 'NONE', 'not compressed'))
|
|
output.writeframes(samples.tobytes())
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
def generate_tone(root: Path, pitch: Fraction) -> dict:
|
|
frequency = 440 * 2 ** (float(pitch - 69) / 12)
|
|
period = Fraction(frequency / SAMPLE_RATE).limit_denominator(SAMPLE_RATE)
|
|
loop_length, cycles = period.denominator, period.numerator
|
|
attack = 12000
|
|
name = hashlib.sha256(f'{PROFILE}:{pitch}'.encode('ascii')).hexdigest()[:24]
|
|
relative = f'audio/{name}.wav'
|
|
harmonics = [h for h in range(1, 7) if frequency * h < SAMPLE_RATE * 0.45]
|
|
|
|
def sample(frame):
|
|
phase = 2 * math.pi * cycles * frame / loop_length
|
|
envelope = min(1, frame / 240) * (0.18 + 0.82 * (1 - min(1, frame / attack)) ** 3)
|
|
return 0.3 * envelope * sum(math.sin(phase * h) / (h * h) for h in harmonics)
|
|
|
|
checksum = write_pcm(root / relative, (sample(i) for i in range(attack + loop_length)))
|
|
return {'id': name, 'path': relative, 'sha256': checksum, 'sample_rate': SAMPLE_RATE, 'frames': attack + loop_length, 'loop_start': attack, 'loop_end': attack + loop_length}
|
|
|
|
def compile_tree(root: Path) -> None:
|
|
tones = {}
|
|
release = {'path': 'audio/release.wav', 'sample_rate': SAMPLE_RATE, 'frames': 2400}
|
|
release['sha256'] = write_pcm(root / release['path'], ((1 - i / 2399) ** 2 for i in range(2400)))
|
|
for path in sorted(root.rglob('*.json')):
|
|
document = json.loads(path.read_text(encoding='utf-8'))
|
|
try:
|
|
guitar = path.relative_to(root).parts[:2] == ('practice', 'chord') or document['staves'][0].get('arrangement_part_type') in ('acoustic', 'electric')
|
|
playback = compile_timeline(document, guitar=guitar)
|
|
used = {}
|
|
for event in playback['events']:
|
|
event['end'] = rational(fraction(event['onset']) + fraction(event.pop('duration')))
|
|
pitch = fraction(event.pop('pitch'))
|
|
if pitch not in tones:
|
|
tones[pitch] = generate_tone(root, pitch)
|
|
tone = tones[pitch]
|
|
event['sample'] = tone['id']
|
|
used[tone['id']] = tone
|
|
playback.update({'samples': list(used.values()), 'release': release, 'profile': PROFILE})
|
|
except PlaybackUnavailable as error:
|
|
playback = {'version': 1, 'available': False, 'reason': str(error)}
|
|
document['playback'] = playback
|
|
path.write_text(json.dumps(document, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
|
|
|
|
def validate_assets(playback: dict, root: Path, cache: dict | None = None) -> None:
|
|
if playback.get('version') != 1 or type(playback.get('available')) is not bool:
|
|
raise ValueError('invalid playback manifest version/availability')
|
|
if not playback['available']:
|
|
if not isinstance(playback.get('reason'), str) or not playback['reason']:
|
|
raise ValueError('unavailable playback must explain why')
|
|
return
|
|
cache = {} if cache is None else cache
|
|
|
|
def check_sample(value):
|
|
name = value['path']
|
|
relative = Path(name)
|
|
if not name.startswith('audio/') or relative.is_absolute() or relative.suffix != '.wav' or any(p in ('', '.', '..') for p in name.split('/')) or '\\' in name or '\x00' in name:
|
|
raise ValueError('invalid generated audio path')
|
|
path = (root / relative).resolve()
|
|
if not path.is_relative_to(root.resolve()):
|
|
raise ValueError('generated audio escapes asset root')
|
|
key = (str(path), value['sha256'], value['frames'], value['sample_rate'])
|
|
if key in cache:
|
|
return cache[key]
|
|
if value['sample_rate'] != SAMPLE_RATE or not 0 < value['frames'] <= 96000:
|
|
raise ValueError('invalid generated audio rate/length')
|
|
if path.stat().st_size != 44 + value['frames'] * 2 or hashlib.sha256(path.read_bytes()).hexdigest() != value['sha256']:
|
|
raise ValueError('generated audio checksum mismatch')
|
|
with wave.open(str(path)) as audio:
|
|
pcm = audio.readframes(value['frames'])
|
|
if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate(), audio.getnframes()) != (1, 2, SAMPLE_RATE, value['frames']) or len(pcm) != value['frames'] * 2:
|
|
raise ValueError('invalid or truncated generated PCM WAV')
|
|
samples = array('h')
|
|
samples.frombytes(pcm)
|
|
if sys.byteorder != 'little':
|
|
samples.byteswap()
|
|
if min(samples) < -32767:
|
|
raise ValueError('generated PCM exceeds unit range')
|
|
cache[key] = samples
|
|
return samples
|
|
|
|
duration = fraction(playback['duration'])
|
|
if duration <= 0 or len(playback['events']) > MAX_EVENTS:
|
|
raise ValueError('invalid playback timeline length')
|
|
release = check_sample(playback['release'])
|
|
if not 1 < playback['release']['frames'] <= SAMPLE_RATE // 2:
|
|
raise ValueError('invalid release envelope length')
|
|
if release[0] != 32767 or release[-1] != 0 or any(a < b for a, b in zip(release, release[1:])):
|
|
raise ValueError('release envelope must decrease from one to zero')
|
|
if len(playback['samples']) > 512 or sum(s['frames'] for s in playback['samples']) > 16 * 1024 * 1024:
|
|
raise ValueError('generated playback exceeds sample memory budget')
|
|
samples = {}
|
|
for sample in playback['samples']:
|
|
if not sample['id'] or sample['id'] in samples or not 0 <= sample['loop_start'] < sample['loop_end'] == sample['frames']:
|
|
raise ValueError('invalid sample identity or loop')
|
|
check_sample(sample)
|
|
samples[sample['id']] = sample
|
|
previous = Fraction()
|
|
identities = set()
|
|
for event in playback['events']:
|
|
onset, end = fraction(event['onset']), fraction(event['end'])
|
|
if event['id'] in identities or event['sample'] not in samples or not previous <= onset < end <= duration:
|
|
raise ValueError('invalid playback event identity, reference or timing')
|
|
identities.add(event['id'])
|
|
previous = onset
|