product-vamp/generator/solo/recognition.py

"""Recognize bounded harmonic relationships in actual sounded music.

No filenames, annotations, learned examples or role exemptions enter here.
Existing expression validators prove each candidate; deterministic set-cover
search chooses a mutually consistent explanation, never edits the notes.
"""
import fractions
import generator.vocabulary
import generator.solo.expressions
import generator.solo.types

def signature(c):
  return (c.kind, c.rule, tuple(n.event_id for n in c.events), c.target.event_id,
    str(c.boundary), c.preparation.event_id if c.preparation else '')

def merge(*groups):
  result = {}
  for group in groups:
    for c in group:
      result.setdefault(signature(c), c)
  return tuple(result.values())

def recognize(notes, harmony, provided=(), *, budget=4096):
  required = {n.event_id for n in notes if not harmony.ordinary(n)}
  fixed = {n.event_id for c in provided for n in c.events}
  if required <= fixed:
    return tuple(provided), ''
  candidates = {}
  single = {}
  by_id = {n.event_id: n for n in notes}
  if len(by_id) != len(notes):
    return (), 'Harmonic analysis requires unique sounding-event IDs'

  def contract(kind, rule, events, target, boundary=None, preparation=None):
    span = harmony.at(boundary if boundary is not None else target.start)
    if not span or span.chord == 'N.C.':
      return None
    identifier = f'inferred:{kind}:{rule}:{events[0].event_id}:{target.event_id}:{boundary}'
    return generator.solo.types.ExpressionContract(identifier, kind, rule, tuple(events), target, span.chord, boundary, preparation)

  def add(c, extras=()):
    if c is None:
      return False
    bundle = merge(extras, (c,))
    covered = frozenset(n.event_id for a in bundle for n in a.events)
    if not (covered & required):
      return False
    combined = merge(provided, bundle)
    _, error = generator.solo.expressions.validate(notes, combined, harmony)
    if error:
      return False
    candidates[tuple(signature(a) for a in bundle)] = (bundle, covered)
    return True

  # First discover independently supported colors and harmonic anticipations.
  # These can also support a later approach's destination or passing endpoint.
  for i, n in enumerate(notes):
    if n.event_id not in required:
      continue
    future = [t for t in notes[i + 1:] if n.end <= t.start <= n.end + fractions.Fraction(generator.vocabulary.data('harmony')['resolution_gap'])
      and 0 < abs(t.pitch.number() - n.pitch.number()) <= 2]
    regions = [(None, harmony.at(n.start))]
    regions.extend((s.start, s) for s in harmony.spans if n.start < s.start < n.end
      and s.start - n.start <= fractions.Fraction(generator.vocabulary.data('harmony')['anticipation_max']))
    for boundary, span in regions:
      if not span or span.chord == 'N.C.':
        continue
      if boundary is not None:
        c = contract('anticipation', 'early_chord_tone', (n,), n, boundary)
        if add(c):
          single.setdefault(n.event_id, []).append((c,))
      for rule in generator.vocabulary.data('harmony')['colors']:
        if not generator.solo.expressions.eligible_color(rule, span, harmony) and boundary is not None:
          continue
        kind = 'anticipated_color' if boundary is not None else 'color'
        for target in future if rule.get('resolve') else [n]:
          c = contract(kind, rule['id'], (n,), target, boundary)
          # Non-anticipated color permission is scoped to the note's initial
          # harmony, even when its resolution occurs after a chord boundary.
          if boundary is None and c:
            c = generator.solo.types.ExpressionContract(c.id, c.kind, c.rule, c.events, c.target, span.chord)
          if add(c):
            single.setdefault(n.event_id, []).append((c,))
    if i + 1 < len(notes):
      target = notes[i + 1]
      for s in harmony.spans:
        if n.start < s.start < n.end:
          for rule in generator.vocabulary.data('harmony').get('suspensions', []):
            add(contract('suspension', rule['id'], (n,), target, s.start))
        elif s.start == n.end == target.start and n.pitch.number() == target.pitch.number():
          add(contract('anticipation', 'early_reattack', (n,), target, s.start))
  for i, target in enumerate(notes):
    targets = [(None, ())]
    for support in single.get(target.event_id, ()):
      c = support[0]
      # An approach into an anticipated color needs both permissions; that
      # combination is not silently interpreted as a chord-tone anticipation.
      if c.kind == 'color':
        targets.append((None, support))
      elif c.kind == 'anticipation':
        targets.append((c.boundary, support))
    for rule in generator.vocabulary.data('lead_ins')['patterns']:
      size = len(rule['semitones'])
      if i < size:
        continue
      ns = notes[i - size:i]
      if not any(n.event_id in required for n in ns):
        continue
      if tuple(n.pitch.number() - target.pitch.number() for n in ns) != tuple(rule['semitones']):
        continue
      preparation = notes[i - size - 1] if i > size and rule['kind'] == 'connection' else None
      for boundary, extras in targets:
        add(contract(rule['kind'], rule['id'], ns, target, boundary, preparation), extras)
    # Delayed approaches may include real space or intervening valid music.
    # They still have an exact semitone target and a finite deadline.
    rule = generator.vocabulary.data('lead_ins')['delayed_approach']
    for n in reversed(notes[:i]):
      if target.start - n.start > fractions.Fraction(rule['deadline']):
        break
      if n.event_id in required and target.pitch.number() - n.pitch.number() in rule['steps']:
        add(contract('delayed_approach', rule['id'], (n,), target))
    if i >= 2:
      before, n = notes[i - 2:i]
      if n.event_id in required:
        for left in [(), *single.get(before.event_id, ())]:
          for right in [(), *single.get(target.event_id, ())]:
            add(contract('passing', generator.vocabulary.data('lead_ins')['passing']['id'], (n,), target, preparation=before), merge(left, right))
        for rule in generator.vocabulary.data('lead_ins')['blues']:
          add(contract('blues', rule['id'], (n,), target, preparation=before))
  options = {identifier: [] for identifier in sorted(required - fixed, key=lambda identifier: (by_id[identifier].start, identifier))}
  for bundle, covered in candidates.values():
    for identifier in covered & options.keys():
      options[identifier].append((bundle, covered))
  for identifier, choices in options.items():
    if not choices:
      n = by_id[identifier]
      return (), f'{identifier} at {n.start}: unsupported harmonic relationship for {n.pitch.label()}'
    choices.sort(key=lambda x: (sum((n.end - n.start for c in x[0] for n in c.events), fractions.Fraction()),
      sum((max(fractions.Fraction(), c.target.start - c.events[-1].end) for c in x[0]), fractions.Fraction()), tuple(signature(c) for c in x[0])))
  visits = 0
  failed = set()

  def solve(chosen, covered):
    nonlocal visits
    remaining = required - covered
    if not remaining:
      return chosen
    key = tuple(sorted(signature(c) for c in chosen))
    if key in failed or visits >= budget:
      return None
    visits += 1
    identifier = min(remaining, key=lambda x: (len(options[x]), by_id[x].start, x))
    for bundle, added in options[identifier]:
      combined = merge(chosen, bundle)
      _, error = generator.solo.expressions.validate(notes, combined, harmony)
      if error:
        continue
      result = solve(combined, covered | added)
      if result is not None:
        return result
    failed.add(key)
    return None

  result = solve(tuple(provided), fixed)
  if result is None:
    return (), 'Harmonic treatment search budget exhausted' if visits >= budget else 'No mutually consistent harmonic treatment coverage'
  return result, ''