Revamp

generator/solo/themes.py

@@ -99,10 +99,34 @@ def feature(reference, actual, *, treatment=None):
     tuple(n.pitch.number()+octave for n in original))
 
 
-def rhythm_windows(library,meter,length,context):
+def rhythmic_purpose(operation, terminal=False):
+  if terminal:return 'arrival'
+  if operation in ('introduce','callback','return','repeat_with_harmonic_adaptation'):return 'entrance'
+  if operation in ('answer','fragment'):return 'answer'
+  return 'connect' if operation=='continue' else 'continue'
+
+
+def rhythm_cost(pattern,length,purpose):
+  """Evaluate a rhythmic job, not resemblance to any target score or density."""
+  if not pattern:return float('inf')
+  gaps=[b[0]-a[1] for a,b in zip(pattern,pattern[1:])]
+  isolated=sum(b-a<=F(1,2) and prev[1]<a and b<nxt[0]
+    for prev,(a,b),nxt in zip(pattern,pattern[1:],pattern[2:]))
+  values={b-a for a,b in pattern}
+  uniform=.15 if len(pattern)>=4 and len(values)==1 else 0.
+  if purpose in ('continue','connect'):
+    # Silence may articulate an idea; disconnected tiny islands do not carry
+    # unfinished motion. No minimum occupancy or chorus attack quota.
+    return isolated+.25*float(sum(gaps,F())/length)+uniform
+  if purpose=='arrival':
+    return isolated+float(max(F(),F(1)- (pattern[-1][1]-pattern[-1][0])))/2+uniform
+  return isolated+uniform
+
+
+def rhythm_windows(library,meter,length,context,purpose='continue'):
   from generator.solo import drafts
   tempo=getattr(context,'tempo',120)
-  key=('theme-rhythm',id(library),meter,length,tempo)
+  key=('theme-rhythm',id(library),meter,length,tempo,purpose)
   if key not in context.pools:
     patterns={}
     bar=F(meter.split('/')[0])*4/int(meter.split('/')[1])
@@ -112,8 +136,8 @@ def rhythm_windows(library,meter,length,context):
       # than accidentally excluding eighths/dotted values by snippet length.
       # Pitch paths and feature contracts span the assembled rhythm's seams.
       split=bar
-      for left,left_id in rhythm_windows(library,meter,split,context):
-        for right,right_id in rhythm_windows(library,meter,length-split,context):
+      for left,left_id in rhythm_windows(library,meter,split,context,'continue' if purpose=='arrival' else purpose):
+        for right,right_id in rhythm_windows(library,meter,length-split,context,purpose):
           combined=(*left,*((a+split,b+split) for a,b in right))
           if len(combined)<=16:patterns.setdefault(combined,left_id+' + '+right_id)
     for identifier,pattern,total in drafts.catalog(library,meter):
@@ -123,22 +147,29 @@ def rhythm_windows(library,meter,length,context):
         selected=tuple((a-offset,b-offset) for a,b in pattern if offset<=a and b<=offset+length)
         if selected and len(selected)<=16:patterns.setdefault(selected,identifier)
     # Authored cells can start late and continue over computational barlines.
-    for identifier,pattern,total in generator.vocabulary.rhythm_cells(tempo):
+    for cell in generator.vocabulary.rhythm_cells(tempo,purpose):
+      identifier,pattern,total=cell.id,cell.pattern,cell.duration
       if total>length:continue
       for offset in (F(),F(1,2),max(F(),length-total)):
         if offset+total>length:continue
         head=tuple((a+offset,b+offset) for a,b in pattern)
-        tails=tuple((a,b) for a,b in next(iter(patterns),()) if a>=offset+total)
-        combined=(*head,*tails)
-        if 2<=len(combined)<=16:patterns.setdefault(combined,'authored:'+identifier)
+        # Keep representative compatible continuations; never let dictionary
+        # insertion order secretly choose the only tail for an authored cell.
+        tails=sequences.diverse(tuple(patterns),lambda p:(rhythm_cost(p,length,purpose),policy.stable_rank('rhythm-tail',p)),
+          lambda p:(len(p),p[-1][1]-p[-1][0]),2) or ((),)
+        for tail in tails:
+          if any(a<offset+total<b for a,b in tail):continue
+          combined=(*head,*((a,b) for a,b in tail if a>=offset+total))
+          if 2<=len(combined)<=16:patterns.setdefault(combined,'authored:'+identifier)
     context.pools[key]=sequences.diverse(tuple(patterns.items()),
-      lambda x:(len(x[0])>16,policy.stable_rank('rhythm-window',x[0])),
+      lambda x:(rhythm_cost(x[0],length,purpose)+(0 if x[1].startswith('authored:') else .05),policy.stable_rank('rhythm-window',x[0])),
       lambda x:(x[1].startswith('authored:'),len(x[0]),x[0][0][0],any(a[1]!=b[0] for a,b in zip(x[0],x[0][1:]))),8)
   return context.pools[key]
 
 
 def patterns(ref, start, finish, rhythms, meter, context, *, home=None, operation='introduce', terminal=False, ending=None):
   length=finish-start; values={}
+  purpose=rhythmic_purpose(operation,terminal)
   if home and operation in ('repeat_with_harmonic_adaptation','return'):
     pattern=tuple((n.start,n.end) for n in home.notes if n.end<=length)
     if pattern:values[pattern]='heard-theme'
@@ -148,12 +179,12 @@ def patterns(ref, start, finish, rhythms, meter, context, *, home=None, operatio
     duration=prefix[-1][1]
     if duration==length:values[prefix]=ref.id
     if duration<length:
-      for tail,label in rhythm_windows(rhythms,meter,length-duration,context):
+      for tail,label in rhythm_windows(rhythms,meter,length-duration,context,purpose):
         values[(*prefix,*((a+duration,b+duration) for a,b in tail))]=ref.id+' + '+label
   else:
-    values.update(rhythm_windows(rhythms,meter,length,context))
+    values.update(rhythm_windows(rhythms,meter,length,context,purpose))
   if home and operation=='answer':
-    for p,label in rhythm_windows(rhythms,meter,length,context):values.setdefault(p,label)
+    for p,label in rhythm_windows(rhythms,meter,length,context,purpose):values.setdefault(p,label)
   if not fixed_rhythm:
     # Develop the heard timing, preserving total duration. Eighth/quarter/dotted
     # alternatives are actual contracts, not failed copies of a rhythmic seed.
@@ -174,7 +205,7 @@ def patterns(ref, start, finish, rhythms, meter, context, *, home=None, operatio
       if len(head)>=len(ref.notes):
         arrivals[(*head,(arrival,ending.hold_until-start))]=label+'; planned sustained close'
     values=arrivals
-  return sequences.diverse(tuple(values.items()),lambda x:(len(x[0])>16,policy.stable_rank(ref.id,x[0])),
+  return sequences.diverse(tuple(values.items()),lambda x:(rhythm_cost(x[0],length,purpose),policy.stable_rank(ref.id,x[0])),
     lambda x:(len(x[0]),tuple(b-a for a,b in x[0][:3]),x[0][0][0]),policy.DEFAULT.theme_rhythm_alternatives)
 
 
@@ -251,10 +282,10 @@ def _prefix_paths(templates,ref,first,harmony,library):
   source_motion=original[-1].pitch.number()-original[0].pitch.number()
   for target in pitches(harmony,end_template,first.number()+source_motion,limit=4):
     template=tuple(dataclasses.replace(n,pitch=first if i==0 else target) for i,n in enumerate(templates[:count]))
-    paths.extend(r.notes for r in routes.candidates(template,{template[-1].event_id},harmony,library,harmony.profile.register,'seed '+ref.id))
+    paths.extend(r.notes for r in routes.candidates(template,{template[-1].event_id},harmony,library,harmony.profile.register,'seed '+ref.id,purpose='entrance'))
   valid=[p for p in paths if not references.feature_error(feature(ref,p),p)]
-  return sequences.diverse(valid,lambda ns:(sum(abs(b.pitch.number()-a.pitch.number()) for a,b in zip(ns,ns[1:])),tuple(n.pitch for n in ns)),
-    lambda ns:sequences.family(tuple(b.pitch.number()-a.pitch.number() for a,b in zip(ns,ns[1:]))),4)
+  return sequences.diverse(valid,lambda ns:(routes.path_cost(ns),policy.stable_rank(ref.id,ns)),
+    lambda ns:routes.path_family(ns,harmony),4)
 
 
 @dataclasses.dataclass(frozen=True)
@@ -267,6 +298,8 @@ class Unit:
   score: float
   source: str
   routes: tuple[types.RouteCandidate,...] = ()
+  start: F = F()
+  end: F = F()
 
 
 def unit_candidates(ref,start,finish,incoming,harmony,library,rhythms,meter,context,*,home=None,operation='introduce',terminal=False,batch=0,ending=None,goal=None):
@@ -317,7 +350,7 @@ def unit_candidates(ref,start,finish,incoming,harmony,library,rhythms,meter,cont
           continue
         tail=(prefix[-1],*templates[count:])
         tail=(*tail[:-1],dataclasses.replace(tail[-1],pitch=target))
-        for route in routes.candidates(tail,{tail[-1].event_id},harmony,library,harmony.profile.register,operation+' '+ref.id)[:3+batch]:
+        for route in routes.candidates(tail,{tail[-1].event_id},harmony,library,harmony.profile.register,operation+' '+ref.id,purpose=rhythmic_purpose(operation,terminal))[:3+batch]:
           proposed.append(((*prefix[:-1],*route.notes),(route,)))
       seen=set()
       for actual,connections in proposed:
@@ -337,7 +370,8 @@ def unit_candidates(ref,start,finish,incoming,harmony,library,rhythms,meter,cont
         score=harmony.profile.cost(profiles.weighted_center(actual))+2*checked.exposure+entry+leaps+phrasing.destination_cost(last.pitch,last.start,last.end,harmony,incoming,goal,terminal)
         anchor=ending or types.Anchor(last.event_id+':arrival',last.start,last.start,last.pitch,harmony.members(last.start,last.end),harmony.profile.register,
           last.end-last.start,'intermediate',harmony.at(last.start).chord,hold_until=last.end)
-        result.append(Unit(actual,contract,ref,operation,anchor,score,label,connections))
+        score+=.25*rhythm_cost(tuple((n.start-start,n.end-start) for n in actual),finish-start,rhythmic_purpose(operation,terminal))
+        result.append(Unit(actual,contract,ref,operation,anchor,score,label,connections,start,finish))
   unique={tuple((n.start,n.end,n.pitch) for n in u.notes):u for u in result}
   selected=sequences.diverse(tuple(unique.values()),lambda u:(u.score,policy.stable_rank(ref.id,u.notes)),
     lambda u:(u.notes[0].pitch.number()//12,u.notes[-1].pitch.number(),tuple(n.end-n.start for n in u.notes[:count])),policy.DEFAULT.theme_unit_alternatives)
@@ -366,12 +400,25 @@ def make_home(unit,harmony,start=None,end=None):
   return dataclasses.replace(home,degree_steps=relative_steps(unit.notes,harmony),signature=home.notes[:len(unit.reference.notes)])
 
 
+def unit_end(source,cursor,home,ending,bar):
+  """Bound search by time without cutting the heard signature at a barline."""
+  length=2*bar
+  if home and home.signature:
+    length=home.signature[-1].end
+    while 0<length<bar:
+      length+=home.signature[-1].end
+    length=min(2*bar,max(bar,length))
+  finish=min(source.end,cursor+length)
+  if source.end-finish<bar+2 or ending.earliest<finish:
+    finish=source.end
+  return finish
+
+
 def phrase_candidates(source,analysis,incoming,part,harmony,library,rhythms,meter,tempo,context,seed,*,stage='elaborated',batch=0,final=False,solo_end=None):
   from generator.solo import leads,realization
   if source.end-source.start<2:
     return short_fragment(source,incoming,part,harmony,tempo,seed,final)
   bar=F(meter.split('/')[0])*4/int(meter.split('/')[1])
-  unit_length=2*bar
   solo_end=solo_end or max(c.span.end for c in harmony.choices)
   head=source_notes(analysis)
   seed_key=('source-seeds',analysis,bar,seed)
@@ -379,17 +426,14 @@ def phrase_candidates(source,analysis,incoming,part,harmony,library,rhythms,mete
   roots=context.pools[seed_key]
   ending_options=phrasing.endings(source,incoming,part,harmony,bar,limit=8*(batch+1),final=final)
   states=[(0.,(),incoming.home,incoming,ending) for ending in ending_options]
-  cursor=source.start
-  while cursor<source.end and states:
-    finish=min(cursor+unit_length,source.end)
-    if cursor==source.start and source.end-source.start>=3*bar:
-      # Selected home rhythms finish as objects, not at the nearest barline.
-      signature=incoming.home.signature if incoming.home else ()
-      if signature and signature[-1].end>unit_length:
-        finish=min(source.end,cursor+signature[-1].end)
-    if source.end-finish<6 or any(ending.earliest<finish for *_,ending in states):finish=source.end
+  while states and any(not units or units[-1].end<source.end for _,units,_,_,_ in states):
     next_states=[]
     for cost,units,home,state,ending in states:
+      cursor=units[-1].end if units else source.start
+      if cursor>=source.end:
+        next_states.append((cost,units,home,state,ending))
+        continue
+      finish=unit_end(source,cursor,home,ending,bar)
       introducing=home is None
       if introducing:
         requests=[(r,'introduce') for r in roots]
@@ -437,19 +481,14 @@ def phrase_candidates(source,analysis,incoming,part,harmony,library,rhythms,mete
           following=dataclasses.replace(state,home=actual_home,last=unit.notes[-1],notes=unit.notes,source_reference=state.source_reference or ref,
             history=history,returns=(*state.returns,*((cursor,) if operation=='return' else ())),
             register_peak=max(state.register_peak,*(n.pitch.number() for n in unit.notes)))
-          next_states.append((cost+unit.score+.8*repetition+appearance,(*units,unit),actual_home,following,ending))
+          next_states.append((cost+float(finish-cursor)*(unit.score+.8*repetition+appearance),(*units,unit),actual_home,following,ending))
     def rank(value):
       cost,units,home,state,ending=value
-      attacks=sum(len(u.notes) for u in units)
-      remaining=max(F(),ending.earliest-finish)
-      possible=attacks+2*float(remaining)+(0 if finish==source.end else 1)
-      expected=float(source.end-source.start)
-      activity_bound=0.0  # Activity is a gesture/difficulty property, not a chorus quota.
-      return cost/max(1,len(units))+activity_bound,policy.stable_rank(seed,tuple(u.notes for u in units))
+      elapsed=units[-1].end-source.start
+      return cost/max(.001,float(elapsed)),policy.stable_rank(seed,tuple(u.notes for u in units))
     states=sequences.diverse(next_states,rank,
       lambda v:(v[2].id,v[4].pitch.number(),v[4].duration,len(v[1][-1].notes),v[1][-1].operation),policy.DEFAULT.theme_continuations)
     if not states:return ()
-    cursor=finish
   result=[]
   for cost,units,home,state,ending in states:
     plain=tuple(n for u in units for n in u.notes)