product-vamp/generator/solo/expressions.py

"""Typed, target-driven harmonic expression over a strictly mapped theme."""
from __future__ import annotations
import dataclasses
import functools
from fractions import Fraction as F
from generator import harmony as chord_model, model, pitch as spelling, vocabulary
from generator.solo import types

@functools.lru_cache(maxsize=512)
def parsed_chord(symbol):
  """Private read-only chord analysis shared by pure expression operations."""
  return chord_model.parse_symbol(symbol)


@functools.lru_cache(maxsize=512)
def degree_spelling(letter,accidental,degree,alter):
  root={'letter':letter,'alter':model.rational(accidental),'octave':4}
  p=spelling.shift_pitch(root,degree-1,chord_model.degree_semitones(degree,alter))
  return types.Pitch(p['letter'],model.alteration(p['alter']),p['octave'])


def quality(chord):
  degrees=chord_model.degrees(chord)
  if degrees.get(3)==0 and degrees.get(7)==-1:return 'dominant'
  if degrees.get(3)==-1:return 'half_diminished' if degrees.get(5)==-1 else 'minor'
  if degrees.get(3)==0 and degrees.get(5)==0:return 'major'
  return 'other'


def degree_pitch(chord,degree,alter,near):
  value=degree_spelling(chord['root']['letter'],model.alteration(chord['root']['alter']),degree,alter)
  return dataclasses.replace(value,octave=value.octave+round((near-value.number())/12))


@functools.lru_cache(maxsize=1024)
def chromatic(target,delta):
  # Chromatic neighbors are spelled relative to their exact destination.
  if delta in (-3,-2,-1,1,2,3):
    step=(1 if delta>0 else -1)*(2 if abs(delta)==3 else 1)
    root={'letter':target.letter,'alter':model.rational(target.alter),'octave':target.octave}
    p=spelling.shift_pitch(root,step,delta)
    return types.Pitch(p['letter'],model.alteration(p['alter']),p['octave'])
  raise ValueError('unsupported approach interval')


def eligible_color(rule,span,harmony):
  # An explicit no-chord span (or missing harmony) has no optional colors.
  # In particular, looking ahead for an anticipation must not try to parse
  # the display label N.C. as a written chord.
  if not span or not span.scale or not span.chord:return False
  key=(tuple(rule['qualities']),rule['degree'],rule['alter'],rule.get('function'),span)
  cache=harmony.color_eligibility
  if key not in cache:cache[key]=_eligible_color(rule,span,harmony)
  return cache[key]


def _eligible_color(rule,span,harmony):
  chord=parsed_chord(span.chord)
  degrees=chord_model.degrees(chord)
  if quality(chord) not in rule['qualities']:return False
  degree=rule['degree']; simple=(degree-1)%7+1
  # Explicit alterations and omissions take precedence over optional colors.
  if any((d['value']-1)%7+1==simple and (d['type']=='subtract' or model.alteration(d['alter'])!=rule['alter']) for d in chord.get('degrees',[])):return False
  if degree in degrees and degrees[degree]!=rule['alter']:return False
  following=next((s for s in harmony.spans if s.start>=span.end and s.chord!=span.chord and s.scale),None)
  root=int(model.NATURAL[chord['root']['letter']]+model.alteration(chord['root']['alter']))%12
  next_chord=parsed_chord(following.chord) if following else None
  next_root=int(model.NATURAL[next_chord['root']['letter']]+model.alteration(next_chord['root']['alter']))%12 if next_chord else None
  function=rule.get('function')
  if function=='predominant_to_dominant' and not (next_chord and quality(next_chord)=='dominant' and (next_root-root)%12==5):return False
  if function=='resolving_dominant' and (next_root is None or (root-next_root)%12!=7):return False
  if function=='tonic_or_predominant':
    tonic=model.parse_pitch_name(span.scale.root)
    tonic_pc=int(model.NATURAL[tonic['letter']]+model.alteration(tonic['alter']))%12
    if root!=tonic_pc and not (next_chord and quality(next_chord)=='dominant' and (next_root-root)%12==5):return False
  return True


def validate(notes,contracts,harmony):
  """Return event permissions only after checking complete actual gestures."""
  actual={n.event_id:n for n in notes};approved={}
  def same(expected):
    n=actual.get(expected.event_id)
    return n and (n.start,n.end,n.pitch)==(expected.start,expected.end,expected.pitch)
  for c in contracts:
    if not c.events or not all(same(n) for n in (*c.events,c.target)):
      return {},c.id+': missing or changed expression event/target'
    if c.preparation and not same(c.preparation):return {},c.id+': changed preparation'
    if any(n.event_id in approved for n in c.events):return {},c.id+': overlapping expression permissions'
    target=c.target; span=harmony.at(c.boundary if c.boundary is not None else target.start)
    if not span or not span.scale or span.chord!=c.chord:return {},c.id+': wrong target harmony'
    if c.kind in ('approach','enclosure','connection'):
      rule=next((r for r in vocabulary.data('lead_ins')['patterns'] if r['id']==c.rule and r['kind']==c.kind),None)
      if not rule or len(c.events)!=len(rule['semitones']):return {},c.id+': unknown approach formula'
      arrival=c.boundary if c.boundary is not None else target.start
      if c.boundary is not None and not any(a.kind=='anticipation' and a.target==target and a.boundary==c.boundary for a in contracts):return {},c.id+': future target lacks anticipation contract'
      planned_color=any(a.kind=='color' and a.target==target and a.events==(target,) for a in contracts)
      if not planned_color and (target.pitch.number()%12 not in harmony.members(arrival,target.end) or target.pitch.number()%12 not in harmony.permitted(arrival,target.end)):return {},c.id+': invalid destination hold'
      gaps=tuple(map(F,rule.get('internal_gaps',['0']*(len(c.events)-1))))
      if len(gaps)!=len(c.events)-1 or any(g<0 for g in gaps):return {},c.id+': invalid internal approach gaps'
      gap=F(rule['gap']);cursor=target.start-gap-sum((F(d) for d in rule['durations']),F())-sum(gaps,F())
      if target.start-cursor>F(vocabulary.data('lead_ins')['max_approach']):return {},c.id+': approach too long'
      for i,(n,step,duration) in enumerate(zip(c.events,rule['semitones'],rule['durations'])):
        if n.start!=cursor or n.end!=cursor+F(duration) or n.pitch.number()!=target.pitch.number()+step:return {},c.id+': wrong register, interval, or timing'
        cursor=n.end+(gaps[i] if i<len(gaps) else 0)
      for index in rule.get('mapped_indices',[]):
        n=c.events[index]
        if n.pitch.number()%12 not in harmony.permitted(n.start,n.end):return {},c.id+': declared diatonic neighbor is outside the mapped collection'
      if any(n.start<c.target.start and n.end>c.events[0].start and n.event_id not in {e.event_id for e in c.events} for n in notes):return {},c.id+': interrupted approach'
      if c.kind=='connection' and (not c.preparation or c.preparation.end!=c.events[0].start or abs(c.preparation.pitch.number()-c.events[0].pitch.number())>2):return {},c.id+': chromatic connection lacks its incoming motion'
    elif c.kind=='delayed_approach':
      rule=vocabulary.data('lead_ins').get('delayed_approach',{})
      if c.rule!=rule.get('id') or len(c.events)!=1:return {},c.id+': unknown delayed approach'
      n,=c.events
      if n.end-n.start>F(rule['max_note']) or not n.end<=target.start<=n.start+F(rule['deadline']):return {},c.id+': expired delayed approach'
      if target.pitch.number()-n.pitch.number() not in rule['steps']:return {},c.id+': wrong delayed target register'
      if target.pitch.number()%12 not in harmony.members(target.start,target.end) or target.pitch.number()%12 not in harmony.permitted(target.start,target.end):return {},c.id+': invalid delayed destination'
    elif c.kind=='color':
      rule=next((r for r in vocabulary.data('harmony')['colors'] if r['id']==c.rule),None)
      if not rule:return {},c.id+': unknown color policy'
      colored=False
      for n in c.events:
        if n.end-n.start>F(rule['max_beats']):return {},c.id+': excessive color duration'
        for region,a,b in harmony.segments(n):
          if n.pitch.number()%12 in region.roles.basic:continue
          if not region.scale or not eligible_color(rule,region,harmony):return {},c.id+': incompatible color harmony/function'
          p=degree_pitch(parsed_chord(region.chord),rule['degree'],rule['alter'],n.pitch.number())
          # The rule retains the functional degree/spelling. Encoded
          # enharmonic notation must not change sounding harmonic validity.
          if p.number()!=n.pitch.number():return {},c.id+': color degree/register mismatch'
          colored=True
          if not rule['outside'] and n.pitch.number()%12 not in region.scale.pitches:return {},c.id+': color requires mapped collection'
      if not colored:return {},c.id+': no interval expresses the declared color'
      last=c.events[-1]
      if rule.get('resolve'):
        if target.event_id==last.event_id or not 0<=target.start-last.end<=F(vocabulary.data('harmony')['resolution_gap']) or not 0<abs(target.pitch.number()-last.pitch.number())<=2 or target.pitch.number()%12 not in harmony.members(target.start,target.end):return {},c.id+': unresolved color'
    elif c.kind=='suspension':
      rule=next((r for r in vocabulary.data('harmony').get('suspensions',[]) if r['id']==c.rule),None)
      if not rule or len(c.events)!=1 or c.boundary is None:return {},c.id+': unsupported suspension'
      held,=c.events
      if not held.start<=c.boundary-F(rule['preparation'])<c.boundary<held.end:
        return {},c.id+': insufficient sounding preparation'
      if held.pitch.number()%12 not in harmony.members(held.start,c.boundary):return {},c.id+': nonchordal preparation'
      if held.end-c.boundary>F(rule['max_delay']) or target.start!=held.end or target.pitch.number()-held.pitch.number() not in rule['steps']:
        return {},c.id+': missed suspension resolution'
      if target.pitch.number()%12 not in harmony.members(target.start,target.end):return {},c.id+': invalid suspension target'
      if held.pitch.number()%12 not in harmony.permitted(held.start,held.end):return {},c.id+': suspension outside mapped collections'
    elif c.kind=='anticipated_color':
      rule=next((r for r in vocabulary.data('harmony')['colors'] if r['id']==c.rule),None)
      if not rule or len(c.events)!=1 or c.boundary!=span.start:return {},c.id+': unsupported anticipated color'
      n,=c.events
      if not n.start<c.boundary<n.end or c.boundary-n.start>F(vocabulary.data('harmony')['anticipation_max']):return {},c.id+': invalid anticipated color interval'
      arrived=dataclasses.replace(n,start=c.boundary)
      target=arrived if c.target==n else c.target
      inner=types.ExpressionContract(c.id+':arrived','color',rule['id'],(arrived,),target,span.chord)
      _,error=validate((arrived,*((target,) if target!=arrived else ())), (inner,),harmony)
      if error:return {},error
    elif c.kind=='anticipation':
      if c.rule not in ('early_chord_tone','early_reattack') or len(c.events)!=1 or span.start!=c.boundary:
        return {},c.id+': unknown anticipation or non-boundary target'
      n,=c.events
      if c.boundary is None or not n.start<c.boundary or c.boundary-n.start>F(vocabulary.data('harmony')['anticipation_max']):return {},c.id+': invalid anticipation interval'
      if c.rule=='early_chord_tone' and (n!=target or not c.boundary<n.end):return {},c.id+': invalid held anticipation'
      if c.rule=='early_reattack' and (n.end!=c.boundary or target.start!=c.boundary or target.event_id==n.event_id or target.pitch.number()!=n.pitch.number()):return {},c.id+': invalid anticipated reattack'
      if target.pitch.number()%12 not in harmony.members(c.boundary,target.end) or target.pitch.number()%12 not in harmony.permitted(c.boundary,target.end):return {},c.id+': invalid future destination'
    elif c.kind=='blues':
      if len(c.events)!=1:return {},c.id+': blues treatment requires one inflected event'
      rule=next((r for r in vocabulary.data('lead_ins')['blues'] if r['id']==c.rule),None)
      n,=c.events
      chord=parsed_chord(span.chord)
      if not rule or n.end-n.start>F(1,2) or n.end!=target.start or target.pitch.number()%12 not in span.roles.basic:return {},c.id+': unresolved blues inflection'
      expected=degree_pitch(chord,rule['degree'],rule['alter'],n.pitch.number())
      resolved=degree_pitch(chord,rule['target_degree'],0,target.pitch.number())
      if n.pitch!=expected or target.pitch!=resolved or target.pitch.number()-n.pitch.number()!=1:return {},c.id+': unsupported blues destination'
      if not c.preparation or c.preparation.end!=n.start or abs(c.preparation.pitch.number()-n.pitch.number())>2:return {},c.id+': unprepared blues gesture'
    else:return {},c.id+': unknown expression kind'
    for n in c.events:
      covered=n.start
      for region,a,b in harmony.segments(n):
        if a!=covered or not region.scale:return {},c.id+': missing harmony'
        covered=b
      if covered!=n.end:return {},c.id+': incomplete harmonic coverage'
      approved[n.event_id]=c
  return approved,''


def validate_frontier(notes,contracts,harmony,until,complete):
  """Validate heard prefixes and bounded future obligations separately.

  Future contract events are used ONLY to establish rule feasibility. They do
  not count as heard resolution; the contract stays pending until its entire
  target hold is present. Missing past events and expired deadlines fail.
  """
  if complete:
    approved,error=validate(notes,contracts,harmony)
    return approved,(),error
  actual={n.event_id:n for n in notes};future=dict(actual);pending=[]
  for c in contracts:
    required=tuple(dict.fromkeys((*c.events,c.target,*((c.preparation,) if c.preparation else ()))))
    incomplete=False
    for expected in required:
      heard=actual.get(expected.event_id)
      if expected.start>=until:
        future[expected.event_id]=expected;incomplete=True
      elif heard and heard.pitch==expected.pitch and heard.start==expected.start and heard.end==min(until,expected.end):
        future[expected.event_id]=expected
        incomplete |= heard.end<expected.end
      elif not heard or (heard.start,heard.end,heard.pitch)!=(expected.start,expected.end,expected.pitch):
        return {},(),c.id+': missing/changed event before resolution deadline'
    if incomplete:pending.append(c)
  approved,error=validate(tuple(sorted(future.values(),key=lambda n:(n.start,n.event_id))),contracts,harmony)
  return {key:c for key,c in approved.items() if key in actual},tuple(pending),error