product-vamp/audition/theory_review.py

"""Explain complete optional solos. No composer calls or production-data writes."""
from __future__ import annotations
import argparse
import collections
import json
from fractions import Fraction as F
from pathlib import Path
from audition import reference_review, solo_compare
from generator import harmony, model


def chord(h):
  if not h or not h['root']:return None
  root=h['root']+('#'*int(F(h['alter'])) if F(h['alter'])>0 else 'b'*-int(F(h['alter'])))
  if h['kind']=='other':
    # The explicit displayed quality is evidence; never substitute a target
    # chord for a missing source chord. Preserve any parse error in the report.
    return harmony.parse_symbol(root+(h['display'] or ''))
  return {'root':{'letter':h['root'].lower(),'alter':model.rational(F(h['alter']))},'kind':h['kind'],
    'degrees':[{'value':int(d['value']),'alter':model.rational(F(d['alter'])),'type':d['type']} for d in h['degrees']]}


def members(h):
  c=chord(h)
  if not c:return {}
  root=model.NATURAL[c['root']['letter']]+model.alteration(c['root']['alter'])
  return {int(root+harmony.degree_semitones(d,a))%12:str(d) for d,a in harmony.degrees(c).items()}


def explain(review, source=None):
  """Observed relationships and hypotheses, not retrospective permissions."""
  notes=solo_compare.events(review);offset=F(review['alignment']['encoded_offset'] or 0)
  hs=[{**h,'time':str(F(h['time'])-offset)} for h in review['harmonies']]
  output=[]
  for i,n in enumerate(notes):
    start,end=F(n['start']),F(n['end']);p=int(F(n['pitch']))
    active=next((h for h in reversed(hs) if F(h['time'])<=start),None)
    future=next((h for h in hs if F(h['time'])>start),None)
    relation=[];uncertainty=[]
    try:
      cm=members(active);fm=members(future)
    except (ValueError,KeyError) as e:cm=fm={};uncertainty.append(str(e))
    if p%12 in cm:relation.append('chord member '+cm[p%12])
    else:relation.append('non-basic pitch: needs collection/color/motion analysis')
    if future and start<F(future['time'])<end and p%12 in fm:
      relation.append('anticipated future chord member '+fm[p%12])
    if i and i+1<len(notes):
      a,b=notes[i-1],notes[i+1];x=p-int(F(a['pitch']));y=int(F(b['pitch']))-p
      if F(a['end'])==start and end==F(b['start']) and abs(x)<=2 and abs(y)<=2:
        relation.append('neighbor motion' if x==-y else 'directional passing motion' if x*y>0 else 'turn')
    for target in notes[i+1:i+4]:
      t=F(target['start']);q=int(F(target['pitch']))
      if t-start>F(3,2):break
      if abs(q-p)==1 and end<=t and t-end<=F(1,2):
        relation.append('semitone approach to '+target['pitch']+' at '+target['start']);break
    if end-start>=2:relation.append('sustained event; '+('breath follows' if i+1<len(notes) and F(notes[i+1]['start'])>end else 'continuation follows'))
    label=[a['text'] for a in review['annotations'] if start<=F(a['time'])-offset<end]
    output.append({**n,'relationships':relation,'annotation':label,'uncertainty':uncertainty})
  def signature(ns):
    origin=F(ns[0]['start']);base=int(F(ns[0]['pitch']))
    return tuple((str(F(n['start'])-origin),str(F(n['end'])-F(n['start'])),int(F(n['pitch']))-base) for n in ns)
  repeats=collections.defaultdict(list)
  for i in range(len(notes)-3):repeats[signature(notes[i:i+4])].append(notes[i]['start'])
  source_matches=[]
  if source:
    original=solo_compare.events(source)
    fingerprints={signature(original[i:i+4]):original[i]['start'] for i in range(len(original)-3)}
    source_matches=[{'solo_start':n['start'],'source_start':fingerprints[signature(notes[i:i+4])]} for i,n in enumerate(notes[:-3]) if signature(notes[i:i+4]) in fingerprints]
  return {'events':output,'relative_four_event_returns':[{'signature':k,'starts':v} for k,v in repeats.items() if len(v)>1],
    'source_four_event_relationships':source_matches,'source_relationship_limit':'Exact relative four-event evidence only; absence does not disprove thematic influence.',
    'interpretation':'Descriptive hypotheses; no label waives production validation. No output from this command is loaded by production.'}


def rule_review(review, choices, offset):
  """Read-only recognition under production policies; never run composition.

  Recognition deliberately need not cover every reference note. Failed or
  ambiguous relationships remain failures, rather than changing a policy.
  """
  import dataclasses
  from generator import vocabulary
  from generator.solo import types, harmonic, expressions
  choices=tuple(dataclasses.replace(c,span=dataclasses.replace(c.span,start=c.span.start-offset,end=c.span.end-offset)) for c in choices)
  h=harmonic.Harmony(choices);ns=[]
  for i,n in enumerate(solo_compare.events(review)):
    p=n['spelling'];ns.append(types.Note(F(n['start']),F(n['end']),types.Pitch(p['step'].lower(),F(p['alter']),p['octave']-1),event_id=str(i)))
  contracts=[];rejections=collections.defaultdict(list)
  def accept(c):
    if expressions.validate(tuple(ns),(*contracts,c),h)[1]:return False
    contracts.append(c);return True
  for n in ns:
    for boundary in (s.start for s in h.spans if n.start<s.start<n.end):
      accept(types.ExpressionContract('recognition:'+n.event_id,'anticipation','early_chord_tone',(n,),n,h.at(boundary).chord,boundary=boundary))
  for i,n in enumerate(ns):
    span=h.at(n.start)
    if not span or not span.scale:continue
    if any(n in c.events for c in contracts):continue
    if n.pitch.number()%12 in span.roles.basic:continue
    for rule in vocabulary.data('harmony')['colors']:
      target=ns[i+1] if rule.get('resolve') and i+1<len(ns) else n
      c=types.ExpressionContract('recognition:'+n.event_id,'color',rule['id'],(n,),target,h.at(target.start).chord if h.at(target.start) else '')
      if accept(c):break
      written=harmony.parse_symbol(span.chord)
      proposed=expressions.degree_pitch(written,rule['degree'],rule['alter'],n.pitch.number())
      if expressions.quality(written) in rule['qualities'] and proposed.number()==n.pitch.number():
        error=expressions.validate(tuple(ns),(*contracts,c),h)[1]
        rejections[n.event_id].append({'rule':rule['id'],'expected_spelling':proposed.label(),'rejection':error})
  for target in ns:
    anticipated=next((c for c in contracts if c.kind=='anticipation' and c.target==target),None)
    span=h.at(anticipated.boundary if anticipated else target.start)
    if not span or not span.scale:continue
    for rule in vocabulary.data('lead_ins')['patterns']:
      start=target.start-F(rule['gap'])-sum(map(F,rule['durations']),F())
      events=tuple(n for n in ns if start<=n.start<target.start)
      if not events:continue
      prior=next((n for n in reversed(ns) if n.end<=start),None)
      c=types.ExpressionContract('recognition:'+target.event_id+':'+rule['id'],rule['kind'],rule['id'],events,target,span.chord,
        boundary=anticipated.boundary if anticipated else None,preparation=prior if rule['kind']=='connection' else None)
      if accept(c):break
  permissions,error=expressions.validate(tuple(ns),tuple(contracts),h)
  if error:raise ValueError(error)
  events=[]
  for n in ns:
    spans=h.segments(n);mapped=bool(spans) and spans[0][1]==n.start and spans[-1][2]==n.end and all(s.scale and n.pitch.number()%12 in s.scale.pitches for s,a,b in spans)
    basic=bool(spans) and all(n.pitch.number()%12 in s.roles.basic for s,a,b in spans)
    permission=permissions.get(n.event_id)
    status='recognized_treatment' if permission else 'ordinary_valid' if mapped and (basic or n.end-n.start<2) else 'unexplained_or_unsupported'
    events.append({'start':str(n.start),'end':str(n.end),'pitch':n.pitch.label(),'mapped_throughout':mapped,
      'chord_member_throughout':basic,'status':status,'rule':permission.rule if permission else None,
      'candidate_rejections':rejections.get(n.event_id,[]) if status=='unexplained_or_unsupported' else [],
      'harmonies':[s.chord for s,a,b in spans],'collections':[s.scale.label() if s.scale else None for s,a,b in spans]})
  return {'events':events,'rules':dict(collections.Counter(c.rule for c in contracts)),
    'statuses':dict(collections.Counter(n['status'] for n in events)),
    'limit':'Conservative recognition, not optimal parsing. Unrecognized events may be valid under a relationship not recognized here. Production harmony comes from the original source, never the target.'}


def main():
  p=argparse.ArgumentParser(description=__doc__)
  p.add_argument('--reference',action='append',required=True,metavar='ROLE=XML')
  p.add_argument('--source',action='append',required=True,metavar='SONG=XML')
  p.add_argument('--generated-dir',type=Path,required=True)
  p.add_argument('--output',type=Path,required=True)
  args=p.parse_args();args.output.mkdir(parents=True,exist_ok=True)
  sources={k:reference_review.read(v,'original') for k,v in (s.split('=',1) for s in args.source)}
  from generator import musicxml
  from generator.scales import timeline, selection, library
  scales=library.load_library(Path('data/scales'));source_maps={}
  for name,path in (s.split('=',1) for s in args.source):
    document=musicxml.load_musicxml(path);staff=document['staves'][0]
    end=sum((model.fraction(m['duration']) for m in staff['measures']),F())
    source_maps[name]=selection.select(timeline.read(staff),timeline.Solo(F(),end,'analysis only'),scales)
  manifest=json.loads((args.generated_dir/'manifest.solo').read_text())
  results={};lines=['# Musical decisions in the four supplied solos','','Sounding pitch is encoded guitar pitch minus one octave. Times are exact quarter beats from the first written harmony; preceding material remains separate. All measures and all sounding events are retained.','',
    'The relationships below are observations/hypotheses, not harmonic permission or a claim of listening. Source comparisons inspect relative pitch AND rhythm; annotations remain verbatim.','']
  for role,path in (s.split('=',1) for s in args.reference):
    ref=reference_review.read(path,role,pitch_convention='guitar-written')
    item=manifest['solos'][role];actual=reference_review.read(args.generated_dir/item['file'],role,item['form_start_measure'])
    analysis=explain(ref,sources[role.split('/')[0]]);comparison=solo_compare.compare(actual,ref)
    source=sources[role.split('/')[0]]
    recognized=rule_review(ref,source_maps[role.split('/')[0]],F(source['alignment']['encoded_offset'] or 0))
    results[role]={'reference':ref,'analysis':analysis,'rule_review':recognized,'generated_analysis':explain(actual),
      'exact_musical_match':comparison['exact_musical_match'],'exact_matching_attacks':comparison['exact_matching_attacks'],
      'different_onset_positions':len(comparison['differences']),'reference_playability':comparison['reference_playability']}
    lines.extend(['## '+role,'',f"Exact output match: **{comparison['exact_musical_match']}**; equal pitch/onset/release attacks: {comparison['exact_matching_attacks']}. This is not a perceptual similarity score.",'',
      '| Original measure | Aligned onset | Sounding pitch / duration | Observed relationship | Original annotation |','|---|---|---|---|---|'])
    for n in analysis['events']:
      raw=n['spelling'];pitch=raw['step']+('#'*int(F(raw['alter'])) if F(raw['alter'])>0 else 'b'*-int(F(raw['alter'])))+str(raw['octave']-1)
      lines.append(f"| {n['measure']} | {n['start']} | {pitch} / {F(n['end'])-F(n['start'])} | {'; '.join(n['relationships'])} | {'; '.join(n['annotation']).replace(chr(10),' / ')} |")
    lines.extend(['',f"Relative four-event returns: {analysis['relative_four_event_returns']}",'',f"Exact relative source relationships: {analysis['source_four_event_relationships']}",'',
      'General rule recognition against ORIGINAL source harmony: '+str(recognized['statuses']),
      'Recognized policies: '+str(recognized['rules']),'',recognized['limit'],'',
      '| Onset | Pitch | Collection-valid throughout | Recognition | Policy / source harmony |','|---|---|---|---|---|'])
    for n in recognized['events']:
      failures='; '.join(r['rule']+': '+r['rejection'].split(': ',1)[-1]+' (expected '+r['expected_spelling']+')' for r in n['candidate_rejections'])
      lines.append(f"| {n['start']} | {n['pitch']} | {n['mapped_throughout']} | {n['status']} | {n['rule'] or ''} / {', '.join(n['harmonies'])} {failures} |")
  (args.output/'theory.json').write_text(json.dumps(results,indent=2))
  (args.output/'theory.md').write_text('\n'.join(lines)+'\n')
  print('Reviewed every sounding event in',len(results),'complete reference solos; no composer invoked.')


if __name__=='__main__':main()