product-vamp/generator/solo/phrases.py

"""Source-related melodic sentences realized as whole pitch/rhythm routes.

Original song analysis and general vocabulary are the only musical inputs.
"""
import collections
import dataclasses
import fractions
import functools
import generator.model
import generator.solo.expressions
import generator.solo.groups
import generator.solo.policy
import generator.solo.rhythm
import generator.solo.sequences
import generator.solo.selection
import generator.solo.types
import generator.vocabulary


@dataclasses.dataclass(frozen=True)
class Idea:
  rhythm: tuple
  intervals: tuple
  source: str

  @property
  def length(self):
    return self.rhythm[-1][1]


def idea(notes, source):
  first = notes[0]
  return Idea(tuple((n.start-first.start,n.end-first.start) for n in notes),
    tuple(n.pitch.number()-first.pitch.number() for n in notes),source)


def source_ideas(analysis):
  found = {}
  for i in range(len(analysis.melody)):
    for count in (3,4,5,6):
      notes = analysis.melody[i:i+count]
      if len(notes)!=count or not 2<=notes[-1].end-notes[0].start<=6:
        continue
      value = idea(notes,'head:'+str(notes[0].start))
      if not generator.solo.groups.valid(value.rhythm):
        continue
      key = (value.rhythm,value.intervals)
      old,number = found.get(key,(value,0)); found[key] = (old,number+1)
  def rank(row):
    value,count = row
    intervals = [b-a for a,b in zip(value.intervals,value.intervals[1:])]
    turns = sum(a*b<0 for a,b in zip(intervals,intervals[1:]))
    durations = len({b-a for a,b in value.rhythm})
    return (-min(count,3),-durations,-min(turns,2),-min(7,max(value.intervals)-min(value.intervals)),value.source)
  return tuple(v for v,_ in sorted(found.values(),key=rank)[:12])


@dataclasses.dataclass(frozen=True)
class Arrival:
  note: generator.solo.types.Note
  anchor: generator.solo.types.Anchor
  contracts: tuple = ()
  resume: fractions.Fraction = fractions.Fraction()


class Phrases:
  def __init__(self,analysis,harmony,sequences,rhythms,start,end,seed,enabled):
    self.analysis,self.harmony = analysis,harmony
    self.start,self.end,self.seed,self.enabled = start,end,seed,enabled
    self.sequences = sequences
    self.ideas = source_ideas(analysis)
    library = generator.solo.rhythm.Rhythms(rhythms,analysis.tempo)
    self.patterns = tuple(library.groups)
    self.authored_ids = frozenset(v[0] for v in self.patterns)
    self.arrivals = tuple(sorted({(cell.pattern[-1][1]-cell.pattern[-1][0],cell.duration-cell.pattern[-1][1])
      for cell in generator.vocabulary.rhythm_cells(analysis.tempo,'arrival') if cell.pattern and cell.pattern[-1][1]-cell.pattern[-1][0]>=1}))
    # Catalog families contribute one peer per rhythmic function, not one
    # vote per file. In particular the large 5/4 family cannot dominate 4/4.
    representatives = {}
    lengths = {b-a for a,b in zip(analysis.boundaries,analysis.boundaries[1:])}
    for values in library.catalog.values():
      for name,pattern,length,purposes in values:
        if length not in lengths or not 2<=len(pattern)<=6:
          continue
        descriptor = generator.solo.groups.descriptor(pattern,length)
        representatives.setdefault(descriptor,(name,pattern,length,purposes))
    self.patterns += tuple(representatives.values())
    self.patterns = tuple(dict((generator.solo.groups.descriptor(p,l),(name,p,l,roles))
      for name,p,l,roles in self.patterns).values())
    self.patterns_by_length = {}
    for value in self.patterns:
      self.patterns_by_length.setdefault(value[2],[]).append(value)
    self.center = harmony.profile.reference or 60
    self.rejections = collections.Counter()
    self.proposals = collections.Counter()

  def choose(self,values,domain,key):
    return min(values,key=lambda x:generator.solo.policy.stable_rank(self.seed+domain,(key,x)))

  def horizon(self,start):
    region = self.harmony.at(start)
    if not region or not region.scale:
      return min(self.end,region.end if region else self.end)
    gap = next((s.start for s in self.harmony.spans if start<s.start<self.end and not s.scale),self.end)
    boundaries = [t-self.analysis.body_offset for t in self.analysis.boundaries if start<t-self.analysis.body_offset<=self.end]
    if not boundaries:
      return self.end
    if start<0:
      return min(self.end,fractions.Fraction())
    candidates = boundaries[1:8] or boundaries
    def cost(end):
      previous = self.harmony.at(end-fractions.Fraction(1,8))
      before = self.harmony.at(previous.start-fractions.Fraction(1,8)) if previous else None
      cadence = False
      if previous and before and previous.scale and before.scale:
        a = generator.solo.expressions.parsed_chord(before.chord)
        b = generator.solo.expressions.parsed_chord(previous.chord)
        root = lambda c:int(generator.model.NATURAL[c['root']['letter']]+generator.model.alteration(c['root']['alter']))%12
        cadence = generator.solo.expressions.quality(a)=='dominant' and (root(a)-root(b))%12==7
      breath = any(n.end==end+self.analysis.body_offset and (i+1==len(self.analysis.melody) or self.analysis.melody[i+1].start>n.end) for i,n in enumerate(self.analysis.melody))
      return (abs(boundaries.index(end)+1-4)-2*int(cadence)-int(breath),end)
    return min(gap,min(candidates,key=cost))

  @functools.lru_cache(maxsize=8192)
  def pitches(self,start,end):
    span = self.harmony.at(start)
    if not span or not span.scale:
      return ()
    permitted = self.harmony.permitted(start,end)
    return tuple(p for index in range(-28,29) if self.harmony.profile.permits((p:=generator.solo.types.scale_pitch(span.scale,index)).number()) and p.number()%12 in permitted)

  @staticmethod
  def note(start,end,pitch,role):
    return generator.solo.types.Note(start,end,pitch,role=role,event_id=f'{start}:{end}:{pitch.number()}')

  @functools.lru_cache(maxsize=4096)
  def event_options(self,start,end,structural,enabled):
    """Validate the immutable harmonic interval once, not at every near pitch."""
    result = []
    span = self.harmony.at(start)
    if not span or not span.scale:
      return ()
    for pitch in self.pitches(start,end):
      note = self.note(start,end,pitch,'arrival' if structural else 'motion')
      if self.harmony.evaluate((note,),complete=True).valid:
        if not structural or pitch.number()%12 in self.harmony.members(start,end):
          result.append((note,()))
    if enabled and end-start>=1:
      for rule in generator.vocabulary.data('harmony')['colors']:
        if rule.get('resolve') or not generator.solo.expressions.eligible_color(rule,span,self.harmony):
          continue
        low,high = self.harmony.profile.register
        for near in range(low,high+12,12):
          pitch = generator.solo.expressions.degree_pitch(generator.solo.expressions.parsed_chord(span.chord),rule['degree'],rule['alter'],near)
          if not self.harmony.profile.permits(pitch.number()):
            continue
          note = self.note(start,end,pitch,'color')
          color = generator.solo.types.ExpressionContract(note.event_id+':'+rule['id'],'color',rule['id'],(note,),note,span.chord)
          if self.harmony.evaluate((note,),expressions=(color,),complete=True).valid:
            result.append((note,(color,)))
    return tuple(dict.fromkeys(result))

  def options(self,start,end,near,structural=False):
    return tuple(v for v in self.event_options(start,end,structural,self.enabled) if abs(v[0].pitch.number()-near)<=12)

  def arrival_options(self,onset,release,harmonic_time,center,effect,resume=None):
    """Choose the sounding target before realizing its incoming gesture."""
    region = self.harmony.at(harmonic_time)
    if not region or not region.scale or not onset<release:
      return ()
    result = []
    if onset<harmonic_time:
      if not self.enabled or region.start!=harmonic_time or not onset<harmonic_time<release:
        return ()
      # A future member need not be legal in the preceding collection. Only
      # the bounded, validated anticipation authorizes that preceding interval.
      candidates = [(p,()) for p in self.pitches(harmonic_time,release)
        if p.number()%12 in self.harmony.members(harmonic_time,release)]
    else:
      candidates = [(n.pitch,cs) for n,cs in self.options(onset,release,center,True)]
    for pitch,support in candidates:
      if abs(pitch.number()-center)>10:
        continue
      note = self.note(onset,release,pitch,'arrival')
      if onset<harmonic_time:
        support = (generator.solo.types.ExpressionContract(note.event_id+':early','anticipation',
          'early_chord_tone',(note,),note,region.chord,boundary=harmonic_time),)
      elif support:
        note = support[0].target
      if not self.harmony.evaluate((note,),expressions=support,complete=True).valid:
        continue
      anchor = generator.solo.types.Anchor('arrival:'+str(onset),onset,onset,pitch,
        frozenset({pitch.number()%12}),self.harmony.profile.register,release-onset,
        'terminal' if effect in ('complete','handoff') else 'intermediate',region.chord,
        False,release,effect)
      result.append(Arrival(note,anchor,support,release if resume is None else resume))
    return tuple(result)

  def cadence(self,time):
    current = self.harmony.at(time)
    previous = self.harmony.at(time-fractions.Fraction(1,8))
    if not current or not previous or not current.scale or not previous.scale or current.chord==previous.chord:
      return False
    a,b = (generator.solo.expressions.parsed_chord(s.chord) for s in (previous,current))
    root = lambda c:int(generator.model.NATURAL[c['root']['letter']]+generator.model.alteration(c['root']['alter']))%12
    return (generator.solo.expressions.quality(a)=='dominant' and
      generator.solo.expressions.quality(b)!='dominant' and (root(a)-root(b))%12==7)

  def destination_choice(self,options,center,domain,key):
    # Each eligible harmonic degree has an octave realization close to the
    # phrase's chosen hand/register region. Register exploration belongs to
    # the phrase choice, not an unrelated octave lottery at every arrival.
    members = {}
    for value in options:
      pc = value.note.pitch.number()%12
      rank = (abs(value.note.pitch.number()-center),value.note.pitch.number())
      if pc not in members or rank<members[pc][0]:
        members[pc] = (rank,value)
    return self.choose(tuple(v for _,v in sorted(members.values(),key=lambda v:v[0])),domain,key)

  def destinations(self,start,end,center,variant):
    """Small connected alternatives in harmonic time, independent of a body."""
    hold,breath = self.choose(self.arrivals,'arrival',(start,variant))
    close = end==self.end or self.choose((True,False),'closure',(start,variant))
    result = []
    if close and end-start>=3:
      release = end-breath
      base = max(start+1,release-hold)
      times = [(base,base)]
      # Near a harmonic change compare anticipation, on-time and delayed
      # entry, rather than retiming an already selected strict melody.
      for span in self.harmony.spans:
        if start+2<=span.start<release-1 and abs(span.start-base)<=2:
          for offset in (fractions.Fraction(-1,2),fractions.Fraction(),fractions.Fraction(1,2)):
            onset = span.start+offset
            if start+1<=onset<release-1:
              times.append((onset,span.start if offset<0 else onset))
      onset,when = self.choose(tuple(dict.fromkeys(times)),'arrival_time',(start,variant))
      effect = 'handoff' if end==self.end else 'complete' if any(self.cadence(s.start) for s in self.harmony.spans if onset<=s.start<release) or self.cadence(when) else 'settle'
      options = self.arrival_options(onset,release,when,center,effect,end)
      if not options:
        return None
      result.append(self.destination_choice(options,center,'destination',(start,variant)))
    limit = result[0].note.start if result else end
    regions = [s for i,s in enumerate(self.harmony.spans) if s.scale and i and
      self.harmony.spans[i-1].chord!=s.chord and start+3<=s.start<limit-3]
    if regions:
      # Local repose is optional; it is not closure at every chord/cell.
      region = self.choose((*regions,None),'local_settle',(start,variant))
      if region:
        offset = self.choose((fractions.Fraction(-1,2),fractions.Fraction(),fractions.Fraction(1,2)), 'local_time',(start,variant))
        onset = region.start+offset
        duration,space = self.choose(self.arrivals,'local_duration',(start,variant))
        release = min(region.end,region.start+duration,limit-1)
        resume = min(release+space,limit-1)
        local_center = round((center+result[0].note.pitch.number())/2) if result else center
        options = self.arrival_options(onset,release,region.start if offset<0 else onset,local_center,'settle',resume)
        if options:
          result.insert(0,self.destination_choice(options,local_center,'local_destination',(start,variant)))
    return tuple(result)

  @functools.lru_cache(maxsize=128)
  def developments(self,motif,purpose):
    if not motif:
      return ()
    if purpose=='return':
      return (motif,)
    values = []
    # The characteristic pitch relationships can survive rhythmic recasting.
    # Source holds remain source music; they do not become obligatory holds
    # at the start of every generated development.
    cursor = fractions.Fraction(); compact = []
    for i,(a,b) in enumerate(motif.rhythm):
      gap = a-motif.rhythm[i-1][1] if i else a
      cursor += min(gap,fractions.Fraction(1,2))
      length = min(b-a,fractions.Fraction(1))
      compact.append((cursor,cursor+length)); cursor += length
    if generator.solo.groups.valid(tuple(compact)):
      values.append(Idea(tuple(compact),motif.intervals,motif.source+':rephrase'))
      if purpose=='extend':
        # Reopen the provisional ending and continue its motion. This is an
        # edit to the existing idea, not a finished idea followed by a new one.
        a,b = compact[-1]
        stop = a+min(b-a,fractions.Fraction(1,2))
        rhythm = (*compact[:-1],(a,stop),(stop,stop+fractions.Fraction(1,2)),(stop+fractions.Fraction(1,2),stop+1))
        step = 2 if motif.intervals[-1]>=motif.intervals[-2] else -2
        values.append(Idea(rhythm,(*motif.intervals,motif.intervals[-1]+step,motif.intervals[-1]+2*step),motif.source+':extend'))
    for name,pattern,length,roles in self.patterns:
      if (len(pattern)==len(motif.intervals) and 1<=length<=4 and
        set(roles)&{'entrance','continue','connect','develop','answer','scalar'}):
        values.append(Idea(pattern,motif.intervals,motif.source+':rhythm:'+name))
    # A full rhythmic recurrence is still a legitimate candidate, not a duty.
    values.append(motif)
    return tuple(dict(((v.rhythm,v.intervals),v) for v in values).values())

  @functools.lru_cache(maxsize=1024)
  def rhythm_candidates(self,cursor,remaining,approaching,fast_allowed):
    desired_roles = {'connect','continue','scalar'} if approaching else {'continue','connect','develop','answer','scalar','arpeggio'}
    eligible = []
    for length,values in sorted(self.patterns_by_length.items()):
      if not min(remaining,2)<=length<=remaining:
        continue
      for value in values:
        name,pattern,length,roles = value
        fast = any((b-a).denominator not in (1,2) for a,b in pattern)
        if (set(roles)&desired_roles and (not fast or fast_allowed) and
          generator.solo.groups.valid(tuple((cursor+a,cursor+b) for a,b in pattern))):
          eligible.append(value)
    return tuple(eligible)

  def scale_step(self,scale,number,step):
    """Continue from sounding pitch, never transfer a degree between maps."""
    try:
      index = generator.solo.types.scale_index(scale,number)
    except ValueError:
      # The preceding attack may belong to the previous collection. Its
      # pitch stays untouched; enter the new one in the requested direction.
      if not step:
        return None
      entries = [generator.solo.types.scale_pitch(scale,i) for i in range(-28,29)]
      entries = [p for p in entries if (p.number()-number)*step>0]
      if not entries:
        return None
      nearest = min(entries,key=lambda p:abs(p.number()-number))
      index = generator.solo.types.scale_index(scale,nearest.number())
      step -= 1 if step>0 else -1
    return generator.solo.types.scale_pitch(scale,index+step)

  def route(self,rhythm,contour,incoming,center,mode='develop',*,degree_path=False,target=None):
    values = self.routes(rhythm,contour,incoming,center,mode,degree_path=degree_path,target=target,limit=1)
    return values[0] if values else None

  def routes(self,rhythm,contour,incoming,center,mode='develop',*,degree_path=False,target=None,limit=3):
    return self._routes(rhythm,contour,incoming,center,mode,degree_path,target,limit,self.enabled)

  @functools.lru_cache(maxsize=1024)
  def _routes(self,rhythm,contour,incoming,center,mode,degree_path,target,limit,enabled):
    """Keep complete paths toward a known future event, not one next step."""
    if not rhythm:
      return ()
    initial = incoming.pitch.number() if incoming else center
    if incoming and mode in ('connect','arpeggio'):
      direction = (1 if target.pitch.number()>=initial else -1) if target else (1 if contour[-1]>=contour[0] else -1)
      center = initial+direction*2
    options = self.options(*rhythm[0],center)
    if mode=='arpeggio':
      options = tuple(v for v in options if v[0].pitch.number()%12 in self.harmony.at(rhythm[0][0]).roles.basic)
    starts = sorted(options,key=lambda v:(abs(v[0].pitch.number()-center)+max(0,abs(v[0].pitch.number()-initial)-5),v[0].pitch.number()))[:6]
    states = [(abs(n.pitch.number()-initial)/4,(n,),c,n.pitch.number()) for n,c in starts]
    def finish_distance(note):
      if target is None:
        return 0
      distance = abs(note.pitch.number()-target.pitch.number())
      return max(0,distance-2)*2+int(distance==0 and note.end==target.start)*2
    for index,(a,b) in enumerate(rhythm[1:],1):
      next_states = []
      for cost,notes,contracts,origin in states:
        wanted = origin+contour[min(index,len(contour)-1)]
        expected = contour[min(index,len(contour)-1)]-contour[min(index-1,len(contour)-1)]
        span = self.harmony.at(a)
        if degree_path:
          if not span or not span.scale:
            continue
          if target is not None:
            # Use octave-aware indices when both pitches belong to this map.
            # Across maps the target remains an actual sounding pitch.
            try:
              displacement = generator.solo.types.scale_index(span.scale,target.pitch.number())-generator.solo.types.scale_index(span.scale,notes[-1].pitch.number())
              step = max(-2,min(2,round(displacement/(len(rhythm)-index+1))))
            except ValueError:
              step = 1 if target.pitch.number()>notes[-1].pitch.number() else -1
            pitch = self.scale_step(span.scale,notes[-1].pitch.number(),step)
          else:
            pitch = self.scale_step(span.scale,notes[-1].pitch.number(),expected)
          if pitch is None:
            continue
          wanted = pitch.number()
          expected = wanted-notes[-1].pitch.number()
        for note,support in self.options(a,b,wanted):
          delta = note.pitch.number()-notes[-1].pitch.number()
          if degree_path:
            if target is None and note.pitch.number()!=wanted:
              continue
            if target is not None:
              try:
                step = generator.solo.types.scale_index(span.scale,note.pitch.number())-generator.solo.types.scale_index(span.scale,notes[-1].pitch.number())
                if abs(step)>2:
                  continue
              except ValueError:
                if abs(delta)>3:
                  continue
          if mode=='arpeggio' and (note.pitch.number()%12 not in span.roles.basic or delta==0):
            continue
          error = abs(note.pitch.number()-wanted)+3*int(delta==0 and expected!=0)+max(0,abs(delta)-7)*2
          if b-a>=1 and note.pitch.number()%12 not in span.roles.basic and not support:
            error += 2
          next_states.append((cost+error,(*notes,note),(*contracts,*support),origin))
      if not next_states:
        return ()
      def rank(state):
        # Admissible reachability pressure uses the actual remaining attacks;
        # a partial short route is never compared to a complete long route.
        distance = max(0,abs(state[1][-1].pitch.number()-target.pitch.number())-3*(len(rhythm)-index)) if target else 0
        return state[0]+distance,tuple(n.pitch.number() for n in state[1])
      states = generator.solo.sequences.diverse(next_states,rank,
        lambda s:(s[1][0].pitch.number(),s[1][-1].pitch.number()),8)
    values = []
    for cost,notes,contracts,_ in sorted(states,key=lambda s:(s[0]+finish_distance(s[1][-1]),tuple(n.pitch.number() for n in s[1]))):
      if self.harmony.evaluate(notes,expressions=contracts,complete=True).valid:
        values.append((notes,contracts))
    return tuple(values[:limit])

  def contour(self,count,direction,mode,motif,time):
    if mode in ('statement','return','develop','answer') and motif:
      values = motif.intervals
      sign = 1
      # Explicit reduction/extension; never cycle a route to fit its rhythm.
      return tuple(sign*values[i] if i<len(values) else sign*values[-1]+direction*2*(i-len(values)+1) for i in range(count))
    paths = self.sequences.connecting(count) if self.sequences else ()
    paths = [p for p in paths if max(map(abs,p.steps))<=(3 if mode=='arpeggio' else 2) and (p.offsets()[-1]>0)==(direction>0) and sum(a*b<0 for a,b in zip(p.steps,p.steps[1:]))<=1]
    span = self.harmony.at(time)
    if paths and span and span.scale:
      path = min(paths,key=lambda p:(abs(abs(p.offsets()[-1])-(count-1)),p.rank()))
      if mode=='connect':
        return path.offsets()
      base = generator.solo.types.scale_pitch(span.scale,0)
      return tuple(generator.solo.types.scale_pitch(span.scale,i).number()-base.number() for i in path.offsets())
    if mode=='connect':
      return tuple(direction*i for i in range(count))
    return tuple(direction*i*(3 if mode=='arpeggio' else 2) for i in range(count))

  def approach(self,arrival,start,variant):
    """Compose a complete optional lead-in before choosing the body rhythm."""
    if not self.enabled:
      return (),()
    family = self.choose(('direct','approach','enclosure'),'approach_family',(start,variant,arrival.note.start))
    if family=='direct':
      return (),()
    rules = [r for r in generator.vocabulary.data('lead_ins')['patterns'] if r['kind']==family and
      (not r.get('difficulty') or self.analysis.tempo<=generator.vocabulary.data('roles')['difficulty'][r['difficulty']+'_max_bpm'])]
    for rule in sorted(rules,key=lambda r:generator.solo.policy.stable_rank(self.seed+'lead_in',(start,variant,r['id']))):
      gaps = tuple(map(fractions.Fraction,rule.get('internal_gaps',[])))
      cursor = arrival.note.start-sum(map(fractions.Fraction,rule['durations']),fractions.Fraction())-sum(gaps,fractions.Fraction())-fractions.Fraction(rule['gap'])
      if cursor<start+1:
        continue
      ns = []
      for i,(step,length) in enumerate(zip(rule['semitones'],rule['durations'])):
        pitch = generator.solo.expressions.chromatic(arrival.note.pitch,step)
        ns.append(self.note(cursor,cursor+fractions.Fraction(length),pitch,family))
        cursor = ns[-1].end+(gaps[i] if i<len(gaps) else 0)
      if any(not self.harmony.profile.permits(n.pitch.number()) for n in ns):
        continue
      early = next((c.boundary for c in arrival.contracts if c.kind=='anticipation'),None)
      c = generator.solo.types.ExpressionContract(arrival.note.event_id+':'+rule['id'],family,rule['id'],tuple(ns),
        arrival.note,arrival.anchor.chord,boundary=early)
      if self.harmony.evaluate((*ns,arrival.note),expressions=(*arrival.contracts,c),complete=True).valid:
        return tuple(ns),(c,)
    return (),()

  def make(self,start,end,incoming,home,history,variant):
    if not any(s.scale and s.start<end and s.end>start for s in self.harmony.spans):
      self.proposals['space'] += 1
      evaluation = generator.solo.selection.evaluate((),start,end,self.harmony)
      return generator.solo.types.Phrase('space:'+str(start),start,end,(),'space','level','open',evaluation=evaluation),()
    motif = idea(home,'realized_theme') if home else self.choose(self.ideas,'source',variant) if self.ideas else None
    purpose = 'statement' if not home else self.choose(('return','develop','answer','extend'),'purpose',(start,variant))
    if purpose=='return' and history and history[-1].purpose=='return':
      purpose = 'develop'
    self.proposals[purpose] += 1
    center = self.center+self.choose((-5,0,5),'register',(start,variant))
    if purpose=='answer' and home:
      movement = home[-1].pitch.number()-home[0].pitch.number()
      center -= 3 if movement>0 else -3 if movement<0 else 0
    if incoming:
      center = max(incoming.pitch.number()-7,min(incoming.pitch.number()+7,center))
    destinations = self.destinations(start,end,center,variant)
    if destinations is None:
      self.rejections['destination_feasibility'] += 1; return None
    ending = destinations[-1].anchor.effect if destinations and destinations[-1].resume==end else 'open'
    terminal = destinations[-1].anchor if ending!='open' else None
    developments = self.developments(motif,purpose)
    notes,contracts,sources,rhythm_groups = [],[],[],[]
    cursor = start
    if start>=0 and purpose!='return':
      cursor += self.choose((fractions.Fraction(),fractions.Fraction(1,2),fractions.Fraction(1)),'entrance',(start,variant))
    signature_notes = (); group_index = 0; fast_groups = 0
    landmarks = (*destinations,*((None,) if not destinations or destinations[-1].resume<end else ()))
    for landmark in landmarks:
      approach,support = self.approach(landmark,cursor,variant) if landmark else ((),())
      stop = approach[0].start if approach else landmark.note.start if landmark else end
      target = approach[0] if approach else landmark.note if landmark else None
      while cursor<stop:
        span = self.harmony.at(cursor)
        if not span or not span.scale:
          cursor = min(stop,span.end if span else stop); continue
        remaining = stop-cursor
        choices = [v for v in developments if v.length<=remaining and generator.solo.groups.valid(tuple((cursor+a,cursor+b) for a,b in v.rhythm))]
        if group_index==0 and choices:
          motif = self.choose(choices,'theme_rhythm',(start,variant))
          pattern,length,source = motif.rhythm,motif.length,motif.source
          mode = purpose if purpose!='extend' else 'develop'
        else:
          # Select complete rhythmic groups by their musical function, keeping
          # ordered timing diversity until pitches and destination are known.
          eligible = self.rhythm_candidates(cursor,remaining,bool(target and remaining<=4),
            fast_groups<generator.vocabulary.data('solo')['subdivision_groups'])
          if eligible:
            # Families have equal standing; the combinatorial catalog's file
            # count never determines how often its material is requested.
            families = {}
            for value in eligible:
              family = 'catalog' if value[0] not in self.authored_ids else 'authored'
              families.setdefault(family,[]).append(value)
            family = self.choose(tuple(families),'rhythm_family',(start,variant,group_index))
            source,pattern,length,_ = self.choose(families[family],'rhythm',(start,variant,group_index))
          else:
            # An explicit moving primitive, never an arbitrary long hold used
            # to fill a leftover span. Tuplets may only enter as whole groups.
            length = min(fractions.Fraction(1,2),remaining)
            pattern,source = ((fractions.Fraction(),length),),'eighth_continuation'
            if not generator.solo.groups.valid(tuple((cursor+a,cursor+b) for a,b in pattern)):
              self.rejections['subdivision_remainder'] += 1; return None
          mode = 'connect' if target and remaining<=4 else self.choose(('connect','develop','arpeggio'),'movement',(start,variant,group_index))
        rhythm = tuple((cursor+a,cursor+b) for a,b in pattern)
        previous = notes[-1] if notes else incoming
        direction = 1 if target and target.pitch.number()> (previous.pitch.number() if previous else center) else -1 if target else self.choose((-1,1),'direction',(start,variant))
        desired = self.contour(len(pattern),direction,mode,motif,cursor)
        # Only the group immediately reaching a target must finish its approach.
        # Earlier material sees the future register but can retain its identity.
        next_target = target if cursor+length==stop else None
        local_center = round((center+target.pitch.number())/2) if target else center
        routes = self.routes(rhythm,desired,previous,local_center,mode,degree_path=mode=='connect',target=next_target)
        if not routes:
          self.rejections['route'] += 1; return None
        ns,cs = routes[variant%len(routes)]
        if not signature_notes and group_index==0 and len(ns)>=3:
          signature_notes = ns
        fast_groups += int(any((b-a).denominator not in (1,2) for a,b in pattern))
        notes.extend(ns); contracts.extend(cs); sources.append(source)
        rhythm_groups.append(generator.solo.groups.RhythmGroup(cursor,cursor+length,'gesture',rhythm,
          cursor%1,source+':'+mode))
        cursor += length; group_index += 1
      if landmark:
        if approach and notes and notes[-1].end==approach[0].start and notes[-1].pitch==approach[0].pitch:
          # Recompose the join as a direct route in another candidate; never
          # keep an accidental duplicated entrance because the label is valid.
          self.rejections['duplicated_approach_entrance'] += 1; return None
        notes.extend(approach); contracts.extend(support)
        note = landmark.note
        if notes and not approach and not landmark.contracts and notes[-1].end==note.start and notes[-1].pitch==note.pitch:
          before = notes[-1]
          used = {n.event_id for c in contracts for n in (*c.events,c.target)}
          held = dataclasses.replace(note,start=before.start,event_id=note.event_id+':carried')
          if before not in signature_notes and before.event_id not in used and self.harmony.evaluate((held,),complete=True).valid:
            notes.pop(); note=held
        notes.append(note); contracts.extend(landmark.contracts)
        rhythm_groups.append(generator.solo.groups.RhythmGroup(
          approach[0].start if approach else note.start,landmark.resume,'arrival',
          tuple((n.start,n.end) for n in (*approach,note)),note.start%1,landmark.anchor.effect))
        sources.append('destination:'+landmark.anchor.effect)
        cursor = landmark.resume
    if not notes:
      return None
    if self.enabled and variant%3:
      # Only genuinely local refinements remain here. Every committed
      # arrival and the treatment's identifying events are protected.
      protected = (*signature_notes,*(n for n in notes if any(n.start<=d.note.start and n.end>=d.note.end and n.pitch==d.note.pitch for d in destinations)))
      notes,contracts = self.expression(tuple(notes),tuple(contracts),protected,start,variant)
    checked = self.harmony.evaluate(notes,expressions=contracts,until=end,complete=True)
    if not checked.valid:
      self.rejections[checked.error] += 1; return None
    anchors = tuple(d.anchor for d in destinations)
    evaluation = generator.solo.selection.evaluate(notes,start,end,self.harmony,incoming=(incoming,) if incoming else (),
      home=home,purpose=purpose,contracts=contracts,center=center,destinations=anchors,history=history,identity=signature_notes)
    if not evaluation.valid:
      self.rejections['arrival_contract'] += 1; return None
    key = generator.solo.policy.stable_rank(self.seed,(start,end,variant))[:12]
    pattern = tuple((n.start-start,n.end-start) for n in notes)
    try:
      generator.solo.groups.describe(pattern,origin=start,source=purpose)
    except ValueError:
      self.rejections['rhythm_group'] += 1; return None
    # Refine diagnostics against the chosen sounding events. Local expression
    # and shared arrival sustains must not leave stale pre-edit rhythm reports.
    rhythm_groups = tuple(dataclasses.replace(g,events=tuple((n.start,n.end) for n in notes
      if g.start<=n.start<g.end)) for g in rhythm_groups)
    direction = 'up' if notes[-1].pitch.number()>notes[0].pitch.number() else 'down' if notes[-1].pitch.number()<notes[0].pitch.number() else 'returning'
    phrase = generator.solo.types.Phrase(key,start,end,tuple(notes),purpose,direction,ending,
      terminal,tuple(contracts),rhythm_groups,'home' if home else key,float(evaluation.total),evaluation,tuple(sources),anchors,signature_notes)
    return phrase,signature_notes

  def expression(self,notes,contracts,protected,start,variant):
    protected_ids = {n.event_id for n in protected}
    used = {n.event_id for c in contracts for n in (*c.events,c.target)}
    targets = [n for n in notes if n.event_id not in used|protected_ids and n.end-n.start>=1 and n.pitch.number()%12 in self.harmony.members(n.start,n.end)]
    if not targets:
      return notes,contracts
    target = self.choose(targets,'expression_target',(start,variant))
    previous = next((n for n in reversed(notes) if n.end<=target.start),None)
    region = self.harmony.at(target.start)
    editable = previous and previous.event_id not in used|protected_ids and previous.end==target.start
    if editable and region.start==target.start and target is not notes[-1] and target.end-target.start>=fractions.Fraction(3,2):
      for rule in generator.vocabulary.data('harmony').get('suspensions',[]):
        if target.pitch.number()-previous.pitch.number() not in rule['steps']:
          continue
        held = dataclasses.replace(previous,end=target.start+fractions.Fraction(1,2),event_id=previous.event_id+':held',role='suspension')
        resolved = dataclasses.replace(target,start=held.end,event_id=target.event_id+':resolved')
        c = generator.solo.types.ExpressionContract(held.event_id,'suspension',rule['id'],(held,),resolved,region.chord,boundary=region.start)
        revised = tuple(held if n is previous else resolved if n is target else n for n in notes)
        if self.harmony.evaluate(revised,expressions=(*contracts,c),complete=True).valid:
          return revised,(*contracts,c)
    # Anticipation changes the actual rhythm before phrase selection. The
    # original committed arrival interval remains sounding in its entirety.
    if editable and region.start==target.start and previous.end-previous.start>=1:
      early = dataclasses.replace(target,start=target.start-fractions.Fraction(1,2),event_id=target.event_id+':early')
      before = dataclasses.replace(previous,end=early.start,event_id=previous.event_id+':short')
      contract = generator.solo.types.ExpressionContract(early.event_id,'anticipation','early_chord_tone',(early,),early,region.chord,boundary=region.start)
      revised = tuple(early if n is target else before if n is previous else n for n in notes)
      if self.harmony.evaluate(revised,expressions=(*contracts,contract),complete=True).valid:
        return revised,(*contracts,contract)
    # Altered colors resolve to an existing destination; the rule, not the
    # annotation or accidental, authorizes their complete exposed interval.
    if editable and variant%2 and previous.end-previous.start<=1:
      span = self.harmony.at(previous.start)
      for rule in generator.vocabulary.data('harmony')['colors']:
        if not rule.get('resolve') or not generator.solo.expressions.eligible_color(rule,span,self.harmony):
          continue
        pitch = generator.solo.expressions.degree_pitch(generator.solo.expressions.parsed_chord(span.chord),rule['degree'],rule['alter'],target.pitch.number())
        color = dataclasses.replace(previous,pitch=pitch,event_id=previous.event_id+':color',role='color')
        c = generator.solo.types.ExpressionContract(color.event_id,'color',rule['id'],(color,),target,region.chord)
        revised = tuple(color if n is previous else n for n in notes)
        if self.harmony.profile.permits(pitch.number()) and self.harmony.evaluate(revised,expressions=(*contracts,c),complete=True).valid:
          return revised,(*contracts,c)
    rules = [r for r in generator.vocabulary.data('lead_ins')['patterns'] if r['kind'] in ('approach','enclosure') and (not r.get('difficulty') or self.analysis.tempo<=generator.vocabulary.data('roles')['difficulty'][r['difficulty']+'_max_bpm'])]
    rules = sorted(rules,key=lambda r:generator.solo.policy.stable_rank(self.seed+'expression',(start,variant,r['id'])))
    for rule in rules:
      gaps = tuple(map(fractions.Fraction,rule.get('internal_gaps',[])))
      cursor = target.start-sum(map(fractions.Fraction,rule['durations']),fractions.Fraction())-sum(gaps,fractions.Fraction())-fractions.Fraction(rule['gap'])
      affected = [n for n in notes if n.start<target.start and n.end>cursor]
      if cursor<start or any(n.event_id in used|protected_ids or n.start<cursor for n in affected):
        continue
      gesture = []
      for i,(delta,length) in enumerate(zip(rule['semitones'],rule['durations'])):
        pitch = generator.solo.expressions.chromatic(target.pitch,delta)
        gesture.append(self.note(cursor,cursor+fractions.Fraction(length),pitch,rule['kind']))
        cursor += fractions.Fraction(length)+(gaps[i] if i<len(gaps) else 0)
      if any(not self.harmony.profile.permits(n.pitch.number()) for n in gesture):
        continue
      before = next((n for n in reversed(notes) if n.end<=gesture[0].start),None)
      if before and (abs(before.pitch.number()-gesture[0].pitch.number())>7 or before.end==gesture[0].start and before.pitch==gesture[0].pitch):
        continue
      revised = tuple(sorted((*[n for n in notes if n not in affected],*gesture),key=lambda n:n.start))
      contract = generator.solo.types.ExpressionContract(target.event_id+':'+rule['id'],rule['kind'],rule['id'],tuple(gesture),target,self.harmony.at(target.start).chord)
      if self.harmony.evaluate(revised,expressions=(*contracts,contract),complete=True).valid:
        return revised,(*contracts,contract)
    return notes,contracts