Files
product-vamp/generator/vocabulary.py
T
khodowany 0e68f5364b Cleanup
2026-10-04 18:48:02 -05:00

105 lines
4.1 KiB
Python

"""Versioned production vocabulary. This module never reads audition inputs."""
from __future__ import annotations
import contextlib
import contextvars
import dataclasses
import functools
import hashlib
import json
from pathlib import Path
from fractions import Fraction as F
ROOT = Path(__file__).resolve().parents[1] / 'data' / 'composition'
ACTIVE = contextvars.ContextVar('composition_vocabulary', default=None)
def load(root=ROOT):
# Content, not just the directory, keys memoization. Long-lived individual
# role builds must observe edited production data without restarting Python.
names = ('roles', 'solo', 'harmony', 'lead_ins', 'rhythms')
return _load(tuple((name, (Path(root)/(name+'.json')).read_text()) for name in names))
@functools.lru_cache(maxsize=16)
def _load(contents):
result = {}
for name, text in contents:
value = json.loads(text)
if value.get('version') != 1 or not value.get('provenance') or not value.get('license'):
raise ValueError(f'{name}: unsupported or unprovenanced production vocabulary')
entries = value.get('cells', value.get('patterns', value.get('colors', value.get('transformations', []))))
if len({v['id'] for v in entries}) != len(entries):
raise ValueError(f'{name}: duplicate vocabulary ID')
result[name] = value
cells=result['rhythms']['cells']+result['solo']['cells']
if len({c['id'] for c in cells})!=len(cells):raise ValueError('Duplicate rhythm identity')
rules=result['solo']
if set(rules['actions'])!={'arpeggio','scalar','neighbor','develop','answer','return','head'}:raise ValueError('Unknown/missing phrase action')
if set(rules['weights'])!={'purpose','continuity','repetition','register'}:raise ValueError('Unknown/missing musical contribution')
if any(v<=0 for v in rules['search'].values()):raise ValueError('Invalid deterministic search budget')
for group in rules['groups']:
if not group['values'] or any(F(v.removeprefix('r:'))<=0 for v in group['values']):raise ValueError('Invalid rhythm group')
if set(group['purpose'])-set(rules['actions']):raise ValueError('Unreachable rhythm purpose')
return result
@functools.lru_cache(maxsize=1)
def _default():
return {'data': load(), 'role': '', 'settings': {}}
def current():
return ACTIVE.get() or _default()
def data(name):
return current()['data'][name]
def role(name=None):
active = current()
name = name or active['role']
return {**active['data']['roles']['roles'].get(name, {}), **active['settings']}
@contextlib.contextmanager
def using(resources, name, settings=None):
token = ACTIVE.set({'data': resources, 'role': name, 'settings': settings or {}})
try:
yield
finally:
ACTIVE.reset(token)
def fingerprint(value):
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
@dataclasses.dataclass(frozen=True)
class RhythmCell:
id: str
pattern: tuple
duration: F
purpose: str
difficulty: str | None = None
subdivision: str = 'binary'
entry: str = 'attack'
exit: str = 'continue'
def rhythm_cells(tempo, purpose=None):
limits = data('roles')['difficulty']
for cell in (*data('rhythms')['cells'],*data('solo')['cells']):
if purpose is not None and cell['purpose'] != purpose:
continue
difficulty = cell.get('difficulty')
if difficulty and tempo > limits[difficulty + '_max_bpm']:
continue
cursor, notes = F(), []
for value in cell['values']:
rest = value.startswith('r:')
duration = F(value[2:] if rest else value)
if duration <= 0:
raise ValueError(cell['id'] + ': nonpositive rhythm duration')
if not rest:
notes.append((cursor, cursor + duration))
cursor += duration
if difficulty and len(notes) > limits['burst_attacks']:
continue
from generator.solo import groups
realized=groups.describe(tuple(notes),source=cell['id'])
family=cell.get('subdivision','binary')
if family=='triplet' and not any(g.family=='triplet' for g in realized):raise ValueError(cell['id']+': missing complete tuplet group')
yield RhythmCell(cell['id'],tuple(notes),cursor,cell['purpose'],difficulty,family,cell.get('entry','attack'),cell.get('exit','continue'))