product-vamp/generator/comping_rhythm.py

"""Complete, source-only comping figures. Time is in rational quarter notes."""
import dataclasses
import fractions

import generator.accompaniment
import generator.model
import generator.solo.analysis
import generator.solo.policy
import generator.vocabulary

@dataclasses.dataclass(frozen=True, order=True)
class Attack:
  lane: str
  start: fractions.Fraction
  end: fractions.Fraction
  harmonic_time: fractions.Fraction | None = None

@dataclasses.dataclass(frozen=True)
class Rhythm:
  events: tuple
  families: tuple
  contributions: tuple
  exceptions: tuple = ()
  pattern_id: str = ''
  @property
  def cost(self):
    return sum(value for _, value in self.contributions)
  @property
  def signature(self):
    return tuple((e.lane, str(e.start), str(e.end - e.start), str(e.harmonic_time)) for e in self.events)

def melody_contexts(document):
  melody = generator.solo.analysis.attacks(document)
  result, cursor = [], fractions.Fraction()
  for m in document['staves'][0]['measures']:
    length = generator.model.fraction(m['duration'])
    result.append(tuple((n.start - cursor, n.end - cursor) for n in melody if n.start < cursor + length and n.end > cursor))
    cursor += length
  return tuple(result)

def spans(measure, prior):
  length = generator.model.fraction(measure['duration'])
  boundaries = sorted({fractions.Fraction(), length, *(generator.model.fraction(h['onset']) for h in measure['harmonies'])})
  result = []
  for a, b in zip(boundaries, boundaries[1:]):
    chord = generator.accompaniment.active_chord(measure, a, prior)
    if result and generator.accompaniment.same_chord(result[-1][2], chord):
      result[-1] = (result[-1][0], b, chord)
    else:
      result.append((a, b, chord))
  return tuple(result)

def flatten(inputs):
  """Harmony/timing adapter only; never concatenate separately chosen rhythms."""
  first = inputs[0][0]
  result = {**first, 'harmonies': []}
  cursor = fractions.Fraction()
  for m, prior, _ in inputs:
    chord = generator.accompaniment.active_chord(m, fractions.Fraction(), prior)
    initial = chord or {'kind': 'none'}
    result['harmonies'].append({**initial, 'onset': generator.model.rational(cursor)})
    result['harmonies'].extend({**h, 'onset': generator.model.rational(cursor + generator.model.fraction(h['onset']))} for h in m['harmonies'] if generator.model.fraction(h['onset']))
    cursor += generator.model.fraction(m['duration'])
  result['duration'] = generator.model.rational(cursor)
  return result

def validate_events(events, part):
  ordered = sorted(events, key=lambda e: e.start)
  for e in ordered:
    duration = e.end - e.start
    if duration <= 0:
      raise ValueError('comping event must have positive sounding duration')
    if part == 'acoustic':
      if e.lane == 'combined':
        if duration <= 1:
          raise ValueError('combined acoustic sustain must exceed one beat')
      elif e.lane not in ('bass', 'upper') or duration not in (fractions.Fraction(1, 2), 1):
        raise ValueError('ordinary acoustic attacks require separate quarter/eighth notes')
    elif e.lane != 'electric':
      raise ValueError('electric comping requires chord events')
  if any(a.end > b.start for a, b in zip(ordered, ordered[1:])):
    raise ValueError('separate comping events cannot overlap; use one combined sustain')

def fit(pattern, measure, prior, part, *, anticipate=False, limit=None):
  """Bind a complete figure to written harmony, with bounded early arrivals."""
  length = generator.model.fraction(measure['duration'])
  limit = length if limit is None else min(length, limit)
  harmony = spans(measure, prior)
  events, exceptions = [], []
  for row in pattern['events']:
    start = fractions.Fraction(row['onset'])
    end = min(start + fractions.Fraction(row['duration']), limit)
    if start >= limit:
      continue
    active = next(((a, b, c) for a, b, c in harmony if a <= start < b), None)
    if active is None or active[2] is None:
      exceptions.append(('space', 'no written harmony at ' + str(start)))
      continue
    _, boundary, chord = active
    target = None
    if anticipate and row.get('anticipate') and row['layer'] in ('upper', 'electric'):
      future = next(((a, b, c) for a, b, c in harmony if a == start + fractions.Fraction(1, 2) and c), None)
      if future and future[0] < end and not generator.accompaniment.same_chord(chord, future[2]):
        target, boundary = future[0], future[1]
    end = min(end, boundary)
    events.append(Attack(row['layer'], start, end, target))
  try:
    validate_events(events, part)
  except ValueError:
    return None
  return tuple(events), tuple(exceptions)

def evaluate(events, part, melody, harmony, lengths):
  activity, support, coverage, articulation = 0., 0., 0., 0.
  cursor = fractions.Fraction()
  for length in lengths:
    local = [e for e in events if cursor <= e.start < cursor + length]
    attacks = [a for a, b in melody if cursor <= a < cursor + length]
    busy = len(attacks) >= length
    sounding = any(e.start < cursor + length and e.end > cursor for e in events)
    target = (1 if busy else 1.5) if part == 'electric' else (2 if busy else 2.5)
    # Sustained occupancy and planned space are not missing attacks.
    activity += .22 * max(0, abs(len(local) - target) - .75) if not (sounding and not local) else 0
    for e in local:
      gap = not any(a <= e.start < b for a, b in melody)
      entrance = any(a == e.start and (i == 0 or a - melody[i - 1][1] >= fractions.Fraction(1, 2)) for i, (a, _) in enumerate(melody))
      support += -.2 if entrance else -.12 if gap else .18 if busy and e.start in attacks else 0
      articulation += .08 if busy and e.end - e.start > 2 else 0
    cursor += length
  for a, b, chord in harmony:
    if chord and b - a >= 1 and not any(e.start < b and e.end > a for e in events):
      coverage += .18  # A preference, never a chord-attack quota.
  divisor = max(1, len(lengths))
  return tuple((k, v / divisor) for k, v in (('activity', activity), ('melody_support', support), ('harmonic_support', coverage), ('articulation', articulation)))

def candidates(measure, prior, part, melody, seed, *, lengths=None, limit=None):
  lengths = lengths or (generator.model.fraction(measure['duration']),)
  policy = generator.vocabulary.accompaniment(part)
  result = []
  for pattern in policy['patterns']:
    if tuple(map(fractions.Fraction, pattern['lengths'])) != tuple(lengths):
      continue
    for anticipate in (False, True):
      fitted = fit(pattern, measure, prior, part, anticipate=anticipate, limit=limit)
      if fitted is None:
        continue
      events, exceptions = fitted
      if anticipate and not any(e.harmonic_time is not None for e in events):
        continue
      contributions = evaluate(events, part, melody, spans(measure, prior), lengths)
      result.append(Rhythm(events, (('pattern', pattern['family']),), contributions, exceptions, pattern['id'] + ('.anticipated' if anticipate else '')))
  if not result:
    raise ValueError(f'measure {measure["number"]}: no {part} pattern for lengths {lengths} and written harmonic/release constraints')
  return representatives(result, policy['pattern_candidates'], seed)

def representatives(candidates, limit, seed):
  unique = {}
  for c in candidates:
    rank = (c.cost, generator.solo.policy.stable_rank(seed, (c.pattern_id, c.signature)))
    if c.signature not in unique or rank < unique[c.signature][0]:
      unique[c.signature] = rank, c
  ordered = [c for _, c in sorted(unique.values(), key=lambda item: item[0])]
  retained, families = [], set()
  for c in ordered:
    if c.families not in families:
      retained.append(c)
      families.add(c.families)
  retained.extend(c for c in ordered if c not in retained)
  return tuple(retained[:limit])

def transition(previous, current):
  if previous is None:
    return 0.
  return .18 * (previous.signature == current.signature) + .04 * (previous.families == current.families)

def select(layers, seed, left=None, right=None):
  """Compare complete two-bar paths, not independently chosen bar fragments."""
  states = [(0., (), (), left)]
  for i, layer in enumerate(layers):
    following = []
    for c in layer:
      options = [(cost + c.cost + transition(previous, c), (*ties, generator.solo.policy.stable_rank(seed, (i, c.pattern_id, c.signature))), (*path, c), c) for cost, ties, path, previous in states]
      following.append(min(options, key=lambda item: item[:2]))
    states = following
  return min(states, key=lambda item: (item[0] + (transition(item[3], right) if right else 0), item[1]))[2]