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.
"""
from fractions import Fraction as F
from generator import vocabulary
from generator.solo import expressions, 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 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=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+F(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<=F(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 vocabulary.data('harmony')['colors']:
        if not 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=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 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 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=vocabulary.data('lead_ins')['delayed_approach']
    for n in reversed(notes[:i]):
      if target.start-n.start>F(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',vocabulary.data('lead_ins')['passing']['id'],(n,),target,preparation=before),merge(left,right))
        for rule in 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),F()),
      sum((max(F(),c.target.start-c.events[-1].end) for c in x[0]),F()),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=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,''