product-vamp/generator/vocabulary.py

"""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
import pathlib
import fractions

ROOT = pathlib.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', 'harmony', 'lead_ins', 'accompaniment')
  return _load(tuple((name, (pathlib.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') != (6 if name == 'accompaniment' else 1) or not value.get('provenance') or not value.get('license'):
      raise ValueError(f'{name}: unsupported or unprovenanced production vocabulary')
    if name == 'accompaniment':
      signatures = {}
      for part in ('electric', 'acoustic'):
        policy = value['instruments'][part]
        if part == 'acoustic' and policy.get('first_attack') != 'bass_per_harmony':
          raise ValueError('acoustic patterns require bass at the first attack of each harmony')
        if part == 'acoustic' and policy.get('attack_assignment') != {'version': 1, 'mode': 'combined_except_eighth_pairs'}:
          raise ValueError('acoustic patterns require automatic eighth-pair assignment')
        groove = policy['groove']
        if any(type(groove[k]) is not int for k in ('cycle_patterns', 'slot_candidates', 'fill_spacing_measures')) or not 1 <= groove['cycle_patterns'] <= 4 or not 4 <= groove['slot_candidates'] <= 8 or groove['fill_spacing_measures'] < 8:
          raise ValueError('invalid bounded accompaniment groove policy')
        identifiers = set()
        signatures[part] = set()
        for pattern in policy['patterns']:
          if pattern['id'] in identifiers or not pattern['family']:
            raise ValueError('duplicate or unnamed accompaniment pattern')
          identifiers.add(pattern['id'])
          lengths = tuple(map(fractions.Fraction, pattern['lengths']))
          if not 1 <= len(lengths) <= 2 or any(n <= 0 for n in lengths):
            raise ValueError('patterns require one or two positive measure lengths')
          meter = pattern.get('meter')
          grouping = tuple(pattern.get('grouping', ()))
          if meter == 'partial':
            if len(lengths) != 1 or grouping:
              raise ValueError('partial patterns require one measure and no grouping')
          elif meter not in ('3/4', '4/4', '5/4') or lengths != (fractions.Fraction(meter[0]),) * 2:
            raise ValueError('full patterns require an explicit supported meter and two complete bars')
          if (meter == '5/4' and grouping not in ((3, 2), (2, 3))) or (meter != '5/4' and grouping):
            raise ValueError('5/4 patterns require a 3+2 or 2+3 grouping')
          # Validate musical events before any measure projection or ties.
          previous, releases = fractions.Fraction(), {}
          onsets = []
          for row in pattern['events']:
            onset, duration = fractions.Fraction(row['onset']), fractions.Fraction(row['duration'])
            if onset < previous or duration <= 0 or onset + duration > sum(lengths):
              raise ValueError('overlapping or out-of-bounds accompaniment pattern')
            if onset.denominator > 2 or duration.denominator > 2:
              raise ValueError('comping vocabulary uses a quarter/eighth grid')
            if part == 'acoustic':
              if row['layer'] != 'acoustic':
                raise ValueError('acoustic pattern layers are assigned from the complete rhythm')
            elif row['layer'] != 'electric':
              raise ValueError('electric pattern must contain electric chord attacks')
            if row.get('anticipate') and row['layer'] not in ('acoustic', 'electric'):
              raise ValueError('only acoustic/electric chords can anticipate')
            # Acoustic overlaps are checked after deterministic string-lane
            # assignment, including neighboring figures in the source form.
            lanes = () if part == 'acoustic' else (row['layer'],)
            if any(releases.get(lane, fractions.Fraction()) > onset for lane in lanes):
              raise ValueError('overlapping accompaniment attacks on the same strings')
            releases.update({lane: onset + duration for lane in lanes})
            previous = onset
            onsets.append(onset)
          if len(lengths) == 2:
            signatures[part].add((meter, grouping, tuple(onsets)))
        families = policy['voicing_families']
        if not families.get('provenance'):
          raise ValueError('guitar voicing families require provenance')
        if part == 'electric':
          if set(families['rank']) != {'drop-two', 'close-position', 'compact-open', 'compact-doubled', 'three-note-shell'} or any(type(v) is not int or not 0 <= v <= 2 for v in families['rank'].values()):
            raise ValueError('invalid electric family ranking')
        else:
          for kind in ('triad', 'tetrad', 'extended'):
            if set(families[kind]) != {'5/1', '5/5', '6/1', '6/5'}:
              raise ValueError('acoustic families require root/fifth bass on both bass strings')
            for orders in families[kind].values():
              if not orders or any(len(o) != 3 or set(o) - {'root', 'fifth', 'guide', 'terminal', 'color'} for o in orders):
                raise ValueError('invalid acoustic upper chord order')
      if signatures['acoustic'] & signatures['electric']:
        raise ValueError('acoustic/electric two-bar onset patterns must be distinct')
      if not 1 <= value['pattern_candidates'] <= 64:
        raise ValueError('accompaniment pattern budget must be 1–64')
    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
  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 accompaniment(part, resources=None):
  value = (resources or current()['data'])['accompaniment']
  return {**{k: v for k, v in value.items() if k != 'instruments'}, **value['instruments'][part]}

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()