khodowany
@@ -0,0 +1,263 @@
|
||||
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
|
||||
Reference in New Issue
Block a user