"""Build-only snapshots and like-for-like listening comparisons; no engine presets."""
from __future__ import annotations
import argparse
import array
import collections
import copy
import fractions
import hashlib
import html
import importlib
import json
import pathlib
import sys
import time
import wave
import generator.arrange.generate
import generator.model
import generator.musicxml
import generator.musicxml_writer
import generator.parts
import generator.playback
import generator.rhythm
import generator.scales.library
import generator.solo.arrangement
import generator.solo.report
import generator.solo.sequences
STAGES = ('baseline', 'A', 'B', 'C', 'D')
def metrics(solo):
notes = solo.notes()
gaps, cursor, run, runs = [], fractions.Fraction(), 0, []
for note in notes:
if note.start > cursor:
gaps.append((cursor, note.start))
if note.end - note.start == fractions.Fraction(1, 2):
run = run + 1 if note.start == cursor else 1
else:
run = 0
runs.append(run)
cursor = note.end
if cursor < solo.end:
gaps.append((cursor, solo.end))
planned = [(p.start, p.end) for phrase in solo.phrases for p in getattr(phrase.plan, 'passages', ()) if p.intent == 'space']
planned.extend((p.plan.end - p.plan.release, p.plan.end) for p in solo.phrases if p.plan.release)
planned.extend((c.span.start, c.span.end) for c in solo.choices if c.scale is None)
unplanned = []
for start, end in gaps:
fragments = [(start, end)]
for a, b in planned:
fragments = [(x, y) for low, high in fragments for x, y in ((low, min(high, a)), (max(low, b), high)) if y > x]
unplanned.extend(b - a for a, b in fragments)
quoted = sum(n.end - n.start for n in notes if n.role == 'quote')
return {
'phrases': len(solo.phrases), 'attacks': len(notes),
'attacks_per_four_beats': float(4 * len(notes) / solo.end),
'sounding_occupancy': float(sum(n.end - n.start for n in notes) / solo.end),
'longest_unplanned_gap_beats': float(max(unplanned, default=0)),
'longest_connected_eighth_run': max(runs, default=0),
'quotation_coverage': float(quoted / solo.end),
'treatments': dict(collections.Counter(d.treatment for p in solo.phrases for d in p.decisions)),
'search': getattr(solo, 'diagnostics', {}),
}
def fingerprint():
digest = hashlib.sha256()
for path in sorted(pathlib.Path('generator').rglob('*.py')):
digest.update(path.as_posix().encode())
digest.update(path.read_bytes())
return digest.hexdigest()
def capture(root, stage):
destination = root / stage
if destination.exists():
raise ValueError(f'{destination}: snapshot already exists; choose a new output root')
rhythms = generator.rhythm.load_library(pathlib.Path('data/rhythms'))
scales = generator.scales.library.load_library(pathlib.Path('data/scales'))
sequences = generator.solo.sequences.load_library(pathlib.Path('data/sequences'))
result = {'stage': stage, 'implementation': fingerprint(), 'songs': {}}
try:
result['policy'] = importlib.import_module('generator.solo.policy').describe()
except ModuleNotFoundError:
result['policy'] = {'baseline': True}
for path in generator.parts.source_paths(pathlib.Path('data/songs')):
started = time.monotonic()
document = generator.parts.prepare_document(generator.musicxml.load_musicxml(path))
plans = generator.arrange.generate.plan_arrangement(document, rhythms)
parts = {part: generator.arrange.generate._arrange_document(document, part, plans) for part in generator.parts.PART_TYPES}
solos = []
generator.solo.arrangement.apply(document, parts, rhythms=rhythms, scales=scales, sequences=sequences, solos=solos)
song = {'source_sha256': hashlib.sha256(path.read_bytes()).hexdigest(), 'parts': parts, 'solos': {}, 'elapsed_seconds': time.monotonic() - started}
regions = generator.solo.arrangement.regions(document)
for solo, index in zip(solos, (3, 4)):
first, count, offset, length = regions[index]
song['solos'][solo.part] = {'first': first, 'count': count, 'offset': str(offset), 'length': str(length), 'tempo': str(solo.tempo), 'metrics': metrics(solo), 'report': generator.solo.report.describe(solo, scales)}
print(path.stem, solo.part, json.dumps(song['solos'][solo.part]['metrics']), flush=True)
result['songs'][path.stem] = song
destination.mkdir(parents=True)
(destination / 'snapshot.json').write_text(json.dumps(result, indent=2) + '\n')
def excerpt(document, first, count):
result = copy.deepcopy(document)
staff = result['staves'][0]
staff['measures'] = staff['measures'][first:first + count]
notes = [n for m in staff['measures'] for v in m['voices'] for n in v['notes']]
identities = {p['id'] for n in notes for p in n['pitches']}
events = {n['id'] for n in notes}
for note in notes:
for pitch in note['pitches']:
for key in ('tie_from', 'tie_to'):
if pitch.get(key) not in identities:
pitch.pop(key, None)
staff['tuplets'] = [t for t in staff.get('tuplets', []) if t['start'] in events and t['end'] in events]
tuplets = {t['id'] for t in staff['tuplets']}
for note in notes:
if 'tuplet_ids' in note:
note['tuplet_ids'] = [i for i in note['tuplet_ids'] if i in tuplets]
offset = sum(generator.model.fraction(m['duration']) for m in document['staves'][0]['measures'][:first])
end = offset + sum(generator.model.fraction(m['duration']) for m in staff['measures'])
for key in ('scale_spans', 'octave_shifts'):
staff[key] = [{**s, 'start': generator.model.rational(max(offset, generator.model.fraction(s['start'])) - offset), 'end': generator.model.rational(min(end, generator.model.fraction(s['end'])) - offset)} for s in staff.get(key, []) if generator.model.fraction(s['start']) < end and generator.model.fraction(s['end']) > offset]
generator.model.validate_document(result)
return result
def mix(documents, tempo, root, cache):
timelines = [generator.playback.compile_timeline(document) for document in documents]
rate = generator.playback.SAMPLE_RATE
factor = fractions.Fraction(60 * rate, tempo)
frames = round(max(generator.model.fraction(t['duration']) for t in timelines) * factor)
output = array.array('f', [0]) * (frames + 2400)
for index, timeline in enumerate(timelines):
gain = 1.0 if index == 0 else 0.55
for event in timeline['events']:
pitch = generator.model.fraction(event['pitch'])
if pitch not in cache:
sample = generator.playback.generate_tone(root, pitch)
with wave.open(str(root / sample['path'])) as audio:
values = array.array('h', audio.readframes(audio.getnframes()))
if sys.byteorder != 'little':
values.byteswap()
cache[pitch] = sample, values
sample, values = cache[pitch]
start = round(generator.model.fraction(event['onset']) * factor)
end = round((generator.model.fraction(event['onset']) + generator.model.fraction(event['duration'])) * factor)
for frame in range(start, min(len(output), end + 2400)):
age = frame - start
position = age if age < sample['loop_end'] else sample['loop_start'] + (age - sample['loop_start']) % (sample['loop_end'] - sample['loop_start'])
release = 1 if frame < end else (1 - (frame - end) / 2399) ** 2
output[frame] += gain * release * values[position] / 32767
return output
def render(root):
snapshots = {stage: json.loads((root / stage / 'snapshot.json').read_text()) for stage in STAGES}
cache, clips, manifest = {}, [], {'sample_rate': generator.playback.SAMPLE_RATE, 'stages': {}, 'clips': []}
for stage, snapshot in snapshots.items():
manifest['stages'][stage] = {k: v for k, v in snapshot.items() if k != 'songs'}
for song_name, song in snapshot['songs'].items():
baseline = snapshots['baseline']['songs'][song_name]
if song['source_sha256'] != baseline['source_sha256']:
raise ValueError(f'{song_name}: source changed between audition stages')
for part, solo in song['solos'].items():
other = 'electric' if part == 'acoustic' else 'acoustic'
first, count = solo['first'], min(8, solo['count'])
foreground = excerpt(song['parts'][part], first, count)
prefix = root / stage / f'{song_name}_{part}'
generator.musicxml_writer.write_musicxml(prefix.with_suffix('.musicxml'), foreground)
prefix.with_suffix('.txt').write_text(solo['report'])
for responsive in [False, True] if stage == 'D' else [False]:
backing = excerpt((song if responsive else baseline)['parts'][other], first, count)
path = prefix.with_name(prefix.name + ('_responsive' if responsive else '_fixed')).with_suffix('.wav')
samples = mix((foreground, backing), fractions.Fraction(solo['tempo']), root, cache)
clips.append((path, samples))
manifest['clips'].append({'path': path.relative_to(root).as_posix(), 'tempo': solo['tempo'], 'responsive_backing': responsive, 'source_sha256': song['source_sha256'], 'metrics': solo['metrics'], 'elapsed_seconds': song['elapsed_seconds']})
# One common attenuation preserves relative loudness and identical backing.
peak = max(abs(value) for _, samples in clips for value in samples)
gain = min(1, 0.95 / peak) if peak else 1
manifest['shared_gain'] = gain
for path, samples in clips:
generator.playback.write_pcm(path, (value * gain for value in samples))
(root / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
write_index(root, snapshots)
print(f'Rendered {len(clips)} comparisons under {root}', flush=True)
def write_index(root, snapshots):
labels = {'baseline': 'Original engine', 'A': 'Scoring repair', 'B': 'Activity shapes', 'C': 'Motifs and retrieval', 'D': 'Phrase lookahead'}
body = ['Solo comparisons
Solo comparisons
Eight bars from the same solo entrance at the source tempo. All stages use the original backing and one shared mix gain. D also includes the normal backing that responds to the new solo.
A: scoring and register. B: activity and breathing. C: transformation contracts, contextual motifs and diverse paths. D: one-phrase lookahead. Expressive playback is deferred.
Metrics below describe the complete solo, not just the excerpt. These are diagnostics; listening and playing determine musical quality. Build fingerprints and diagnostics.
']
for name, song in snapshots['D']['songs'].items():
for part in song['solos']:
body.append('