Simplification
This commit is contained in:
khodowany
@@ -104,7 +104,7 @@ def generate_assets(args: argparse.Namespace) -> None:
|
||||
export_tree(staged)
|
||||
generator.score.validate_generated_tree(staged)
|
||||
if args.check:
|
||||
generator.summary.validate_expected_performance_summary(data_dir / 'expected' / 'performance_summary.json', staged / 'songs')
|
||||
generator.summary.validate_expected_performance_summary(pathlib.Path(__file__).resolve().parents[1] / 'tests' / 'fixtures' / 'performance_summary.json', staged / 'songs')
|
||||
report('compiling playback')
|
||||
generator.playback.compile_tree(staged)
|
||||
report('engraving pages')
|
||||
|
||||
@@ -24,9 +24,10 @@ def compose(analysis,measures,part,end,*,scales,sequences,rhythms,initial=None,e
|
||||
if placed is None:raise construction.SearchExhausted('selected melody has no complete legal guitar path')
|
||||
by_id={n.event_id:n for n in placed[1]}
|
||||
phrases=tuple(dataclasses.replace(p,notes=tuple(by_id[n.event_id] for n in p.notes)) for p in phrases)
|
||||
result=types.Solo(part,choices,phrases,analysis.tempo,end,{'construction':'complete_phrase_frontiers','version':policy.DEFAULT.composition_version,
|
||||
result=types.Solo(part,choices,phrases,analysis.tempo,end,{'construction':'source_melodic_phrases','version':policy.DEFAULT.composition_version,
|
||||
'construction_seconds':time.monotonic()-began,**builder.timings,'operation_coverage':{k:dict(v) for k,v in builder.coverage.items()},
|
||||
'frontiers':builder.frontiers,'completed_candidates':[{'rank':i+1,'evaluation':dataclasses.asdict(s.evaluation),'events':[(str(n.start),str(n.end),n.pitch.number()) for n in s.notes]} for i,s in enumerate(builder.completed)],'rejections':dict(builder.failures),'decisions':builder.decisions,'expressions_enabled':expressions_enabled},profile,start)
|
||||
'whole_stream_diagnostic':dataclasses.asdict(builder.winner_evaluation),
|
||||
'frontiers':builder.frontiers,'completed_candidates':[{'rank':i+1,'evaluation':{'selection_total': str(s.total / (end-start)), 'phrases': [dataclasses.asdict(p.evaluation) for p in s.phrases]},'events':[(str(n.start),str(n.end),n.pitch.number()) for n in s.notes]} for i,s in enumerate(builder.completed)],'rejections':dict(builder.failures),'decisions':builder.decisions,'expressions_enabled':expressions_enabled},profile,start)
|
||||
validate(result)
|
||||
return result
|
||||
|
||||
|
||||
+66
-251
@@ -1,268 +1,83 @@
|
||||
"""Bounded search of joint musical actions; one normal production path.
|
||||
|
||||
Selection compares several histories at common exact-time frontiers. Actions
|
||||
are indivisible; crossing a reporting window does not restart their contracts.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
from collections import Counter
|
||||
"""Bounded comparison of completed source-related phrases."""
|
||||
import collections
|
||||
import dataclasses
|
||||
from fractions import Fraction as F
|
||||
import fractions
|
||||
import time
|
||||
import heapq
|
||||
import itertools
|
||||
import functools
|
||||
import generator.solo.phrases
|
||||
import generator.solo.selection
|
||||
import generator.solo.sequences
|
||||
import generator.solo.policy
|
||||
import generator.vocabulary
|
||||
|
||||
from generator import vocabulary
|
||||
from generator.solo import types, groups, operators, selection, sequences, scoring
|
||||
|
||||
class SearchExhausted(ValueError):pass
|
||||
class SearchExhausted(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class State:
|
||||
total: F = F()
|
||||
recent: tuple = ()
|
||||
phrases: tuple = ()
|
||||
home: tuple = ()
|
||||
last_kind: str = ''
|
||||
parent: object = None
|
||||
action: object = None
|
||||
evaluation: object = None
|
||||
tie: tuple = ()
|
||||
pending: tuple = ()
|
||||
alternatives: tuple = ()
|
||||
actions: tuple = ()
|
||||
notes: tuple = ()
|
||||
full_evaluation: object = None
|
||||
total: fractions.Fraction = fractions.Fraction()
|
||||
|
||||
@property
|
||||
def notes(self):
|
||||
return tuple(n for p in self.phrases for n in p.notes)
|
||||
|
||||
@functools.cached_property
|
||||
def music_key(self):return tuple((n.start,n.end,n.pitch.number()) for n in self.notes)
|
||||
@functools.cached_property
|
||||
def path_key(self):return tuple(a.key for a in self.actions)
|
||||
|
||||
class Builder:
|
||||
def __init__(self,analysis,measures,harmony,sequences,rhythms,end,seed,enabled=True,*,start=F()):
|
||||
self.analysis=analysis;self.harmony=harmony;self.start=start;self.end=end
|
||||
self.operators=operators.Operators(harmony,analysis,end,start,enabled,sequences,rhythms)
|
||||
self.selector=selection.Selector(harmony,analysis,harmony.profile.reference or 60)
|
||||
self.rules=vocabulary.data('solo')['operators'];self.decisions=[];self.failures=Counter();self.coverage={};self.timings=Counter()
|
||||
self.observe=None # Optional read-only diagnostic; its return is ignored.
|
||||
self.successor_cache={};self.retrieval_at=None;self.retrieval_weights=None
|
||||
def __init__(self,analysis,measures,harmony,sequences,rhythms,end,seed,enabled=True,*,start=fractions.Fraction()):
|
||||
self.analysis,self.harmony = analysis,harmony
|
||||
self.start,self.end,self.seed = start,end,seed
|
||||
self.phrases = generator.solo.phrases.Phrases(analysis,harmony,sequences,rhythms,start,end,seed,enabled)
|
||||
self.rules = generator.vocabulary.data('solo')['search']
|
||||
self.timings = {}; self.frontiers = []; self.coverage = collections.defaultdict(collections.Counter); self.decisions = []
|
||||
self.failures = self.phrases.rejections
|
||||
|
||||
def candidates(self,at,state):
|
||||
candidates=self.operators.proposals(at)
|
||||
if state.home and state.last_kind!='return':
|
||||
candidates+=self.operators.related(at,state.home,'return')+self.operators.related(at,state.home,'answer')
|
||||
# Retrieval never compares cost rates of differently sized actions. Each
|
||||
# endpoint has its own ranking; round-robin the endpoints before cutoff.
|
||||
context=(tuple(n.pitch.number() for n in state.recent[-3:]),state.recent[-1].end if state.recent else None)
|
||||
settled=state.recent[-1].end-state.recent[-1].start>=F(3,2) if state.recent else None
|
||||
if at!=self.retrieval_at or self.retrieval_weights is not self.selector.weights:
|
||||
self.successor_cache.clear();self.retrieval_at=at;self.retrieval_weights=self.selector.weights
|
||||
# Cache only endpoint retrieval, whose sufficient incoming state is three
|
||||
# pitches, release, rest function and pending contracts. Home-specific
|
||||
# actions change the immutable catalog identity; full evaluation below
|
||||
# still uses the actual complete home and recent history on every call.
|
||||
key=(id(candidates),context,settled,state.pending,self.rules['per_state'])
|
||||
cached=self.successor_cache.get(key)
|
||||
if cached is not None and cached[0] is candidates:
|
||||
self.timings['retrieval_cache_hits']+=1;retained=cached[1]
|
||||
else:
|
||||
self.timings['retrieval_cache_misses']+=1;buckets={}
|
||||
for order,a in enumerate(candidates):
|
||||
if not self.pending_compatible(state,a):continue
|
||||
cost=self.selector.fast_cost(a,state.recent,context=context)
|
||||
buckets.setdefault(a.end,[]).append((cost,order,a))
|
||||
ranked=[]
|
||||
for end,rows in sorted(buckets.items()):
|
||||
heapq.heapify(rows);ranked.append(rows)
|
||||
retained=[]
|
||||
while ranked and len(retained)<self.rules['per_state']:
|
||||
again=[]
|
||||
for rows in ranked:
|
||||
while rows:
|
||||
cost,_,action=heapq.heappop(rows)
|
||||
if self.operators.is_valid(action):
|
||||
retained.append((cost,action));break
|
||||
if rows:again.append(rows)
|
||||
if len(retained)==self.rules['per_state']:break
|
||||
ranked=again
|
||||
retained=tuple(retained)
|
||||
self.successor_cache[key]=(candidates,retained) # Retain object; no id reuse.
|
||||
if len(self.successor_cache)>64:self.successor_cache.pop(next(iter(self.successor_cache)))
|
||||
for _,a in retained:self.operators.coverage['retained:'+a.kind]+=1
|
||||
if self.observe:self.observe('successors',at,state,tuple(candidates),tuple(a for _,a in retained))
|
||||
return tuple((self.selector.evaluate(a,state.recent,state.home,state.last_kind),a) for _,a in retained)
|
||||
|
||||
def pending_compatible(self,state,action):
|
||||
for contract in state.pending:
|
||||
target=contract.target
|
||||
if action.start<=target.start<action.end:
|
||||
# Reject a missed committed attack/hold before revalidating all the
|
||||
# harmonic relationships. Partial target evidence remains possible.
|
||||
if not any(n.event_id==target.event_id and n.start==target.start
|
||||
and n.end==min(action.end,target.end) and n.pitch==target.pitch for n in action.notes):return False
|
||||
return True
|
||||
|
||||
def advance(self,state,evaluation,action):
|
||||
if not self.pending_compatible(state,action):return None
|
||||
obligations=(*state.pending,*action.contracts);pending=()
|
||||
if not state.pending:
|
||||
pending=tuple(c for c in action.contracts if c.target.end>action.end)
|
||||
elif obligations:
|
||||
heard={n.event_id:n for c in state.pending for n in c.events if n.start<action.start}
|
||||
heard.update({n.event_id:n for n in action.notes})
|
||||
check=self.harmony.evaluate(tuple(sorted(heard.values(),key=lambda n:n.start)),
|
||||
expressions=obligations,until=action.end,complete=action.end==self.end)
|
||||
if not check.valid:return None
|
||||
pending=check.pending
|
||||
recent=(*state.recent,*action.notes)[-8:]
|
||||
notes=types.sounding_notes((*state.notes,*action.notes))
|
||||
actions=(*state.actions,action)
|
||||
# A complete opening, rather than the first four eligible notes, owns the
|
||||
# theme. Keep it separate from the small context used for retrieval.
|
||||
home=state.home
|
||||
boundaries=[x-self.analysis.body_offset for x in self.analysis.boundaries if x-self.analysis.body_offset>self.start]
|
||||
opening_end=boundaries[min(3,len(boundaries)-1)] if boundaries else self.end
|
||||
if not home and action.end>=opening_end and not pending and len(notes)>=3:
|
||||
# The same sounded opening has the same home identity regardless of
|
||||
# whether its last generating action extends beyond this source window.
|
||||
home=tuple(n for n in notes if n.start<opening_end)
|
||||
if evaluation is None:
|
||||
return State(recent=recent,home=home,last_kind=action.kind,parent=state,action=action,pending=pending,actions=actions,notes=notes)
|
||||
full=self.selector.extend(state.full_evaluation,state.notes,action,home,old_home=state.home,start=self.start,
|
||||
contracts=tuple(dict.fromkeys(c for a in actions for c in a.contracts)))
|
||||
return State(full.total,recent,home,action.kind,state,action,full,(),pending,(),actions,notes,full)
|
||||
|
||||
@staticmethod
|
||||
def identity(state):
|
||||
# Keep rhythmic histories, home identity, and obligations, not just pitch.
|
||||
return (state.music_key,
|
||||
selection.signature(state.home),state.pending)
|
||||
def candidates(self,start,end,state):
|
||||
incoming = state.notes[-1] if state.notes else None
|
||||
values = {}
|
||||
for variant in range(self.rules['alternatives']):
|
||||
value = self.phrases.make(start,end,incoming,state.home,state.phrases,variant)
|
||||
if value is not None:
|
||||
phrase,introduced = value
|
||||
self.coverage[phrase.purpose]['valid'] += 1
|
||||
key = tuple((n.start,n.end,n.pitch.number()) for n in phrase.notes)
|
||||
values.setdefault(key,value)
|
||||
for phrase,_ in values.values():
|
||||
self.coverage[phrase.purpose]['distinct'] += 1
|
||||
return tuple(values.values())
|
||||
|
||||
def rank(self,state):
|
||||
return (state.total,state.music_key,state.path_key)
|
||||
|
||||
def diversity(self,state):
|
||||
# Protect alternative *openings* before a home theme has been selected.
|
||||
# Recent-note diversity alone lets every history inherit the same opening.
|
||||
first_bar=next((x-self.analysis.body_offset for x in self.analysis.boundaries if x-self.analysis.body_offset>max(F(),self.start)),self.end)
|
||||
opening=tuple(itertools.takewhile(lambda n:n.start<first_bar,state.notes))
|
||||
return (selection.signature(opening),opening[0].pitch.number()//12 if opening else None)
|
||||
|
||||
def retain(self,states,at):
|
||||
normalized=[]
|
||||
for s in states:
|
||||
if s.actions and (s.evaluation is None or F(s.evaluation.denominators['span'])!=at-self.start):
|
||||
e=self.frontier_score(s,at);s=dataclasses.replace(s,total=e.total,evaluation=e)
|
||||
normalized.append(s)
|
||||
states=normalized
|
||||
unique={}
|
||||
for s in states:
|
||||
key=self.identity(s)
|
||||
if key not in unique or self.rank(s)<self.rank(unique[key]):unique[key]=s
|
||||
values=sorted(unique.values(),key=self.rank)
|
||||
# Half by score, half by realized pitch/rhythm/obligation diversity. Labels
|
||||
# do not earn a bonus, nor do duplicate derivations receive extra votes.
|
||||
limit=self.rules['histories'];best=values[:max(1,limit//2)]
|
||||
rest=sequences.diverse(values[len(best):],self.rank,self.diversity,limit-len(best))
|
||||
result=tuple(sorted((*best,*rest),key=self.rank))
|
||||
self.frontiers.append({'time':str(at),'proposed':len(states),'distinct':len(unique),'retained':len(result),
|
||||
'best':str(result[0].total) if result else None})
|
||||
if self.observe:self.observe('frontier',at,None,tuple(states),result)
|
||||
return result
|
||||
|
||||
def frontier_score(self,state,until):
|
||||
# The already chosen action can continue beyond this computational cut.
|
||||
# Only its heard prefix participates in an equal-time comparison. The
|
||||
# original notes, full route and obligations remain intact in the state.
|
||||
if until==state.action.end and state.full_evaluation is not None:return state.full_evaluation
|
||||
a=state.action
|
||||
ns=tuple(dataclasses.replace(n,end=min(n.end,until)) if n.end>until else n for n in a.notes if n.start<until)
|
||||
clipped=dataclasses.replace(a,end=until,notes=ns)
|
||||
home=state.home if state.home and state.home[-1].end<=until else ()
|
||||
parent=state.parent
|
||||
return self.selector.extend(parent.full_evaluation,parent.notes,clipped,home,old_home=parent.home,start=self.start,
|
||||
contracts=tuple(dict.fromkeys(c for action in state.actions for c in action.contracts)))
|
||||
return state.total,generator.solo.policy.stable_rank(self.seed,tuple((n.start,n.end,n.pitch.number()) for n in state.notes))
|
||||
|
||||
def compose(self):
|
||||
self.decisions=[];self.frontiers=[];began=time.monotonic()
|
||||
self.successor_cache.clear()
|
||||
self.timings['retrieval_cache_hits']=self.timings['retrieval_cache_misses']=0
|
||||
states=(State(),);at=self.start
|
||||
while at<self.end:
|
||||
# Source quarter-time frontiers are computational, not phrase endings.
|
||||
# Tuplets, delayed approaches and sustained notes retain their full tails.
|
||||
until=min(self.end,F(at.numerator//at.denominator+1))
|
||||
agenda={};ready=[];queue=[];queued=set()
|
||||
began = time.monotonic(); self.frontiers = []; self.decisions = []
|
||||
states = (State(),); start = self.start
|
||||
while start<self.end:
|
||||
end = self.phrases.horizon(start)
|
||||
successors = []
|
||||
for state in states:
|
||||
cursor=state.action.end if state.action else self.start
|
||||
if cursor>=until:ready.append(state)
|
||||
else:
|
||||
agenda.setdefault(cursor,[]).append(state)
|
||||
if cursor not in queued:heapq.heappush(queue,cursor);queued.add(cursor)
|
||||
while queue:
|
||||
cursor=heapq.heappop(queue)
|
||||
partial=self.retain(agenda.pop(cursor),cursor)
|
||||
for state in partial:
|
||||
for evaluation,action in self.candidates(cursor,state):
|
||||
next_state=self.advance(state,evaluation,action)
|
||||
if next_state is None:continue
|
||||
if action.end>=until:ready.append(next_state)
|
||||
else:
|
||||
agenda.setdefault(action.end,[]).append(next_state)
|
||||
if action.end not in queued:heapq.heappush(queue,action.end);queued.add(action.end)
|
||||
# This exact-time frontier will never be expanded again. Keep chosen
|
||||
# actions in immutable histories, not tens of thousands of rejected
|
||||
# concrete actions from obsolete frontiers (especially across workers).
|
||||
self.operators.cache.clear();self.operators.validity_cache.clear()
|
||||
self.selector.static.clear();self.selector.cost_static.clear()
|
||||
if not ready:raise SearchExhausted(f'No continuation at {until} within deterministic search budget')
|
||||
scored=[]
|
||||
for state in ready:
|
||||
evaluation=self.frontier_score(state,until)
|
||||
scored.append(dataclasses.replace(state,total=evaluation.total,evaluation=evaluation))
|
||||
states=self.retain(scored,until);at=until
|
||||
completed=tuple(s for s in states if not s.pending)
|
||||
if not completed:raise SearchExhausted('No complete phrase within deterministic frontier budget; not a proof of musical impossibility')
|
||||
completed=tuple(sorted(completed,key=self.rank))
|
||||
self.completed=(completed[0],*sequences.diverse(completed[1:],self.rank,self.diversity,self.rules['completed']-1))
|
||||
final=completed[0]
|
||||
self.timings['action_search_seconds']=time.monotonic()-began
|
||||
self.timings['frontiers']=len(self.frontiers)
|
||||
path=[];state=final
|
||||
while state.action:path.append(state);state=state.parent
|
||||
path.reverse();phrases=[];pending=[]
|
||||
self.selected_actions=final.actions
|
||||
boundaries=[x-self.analysis.body_offset for x in self.analysis.boundaries]
|
||||
checkpoints=set(boundaries[4::4])
|
||||
for state in path:
|
||||
action=state.action;pending.append(state)
|
||||
self.operators.coverage['selected:'+action.kind]+=1
|
||||
self.decisions.append({'start':str(action.start),'end':str(action.end),'selected':action.kind,
|
||||
'evaluation':dataclasses.asdict(state.evaluation),'prefix_total':state.total})
|
||||
# Reporting partitions never invent a cadence or split an active action.
|
||||
boundary=any(action.start<x<=action.end for x in checkpoints)
|
||||
if (boundary and not state.pending) or action.end==self.end:
|
||||
phrases.append(self.phrase(pending,len(phrases)));pending=[]
|
||||
for name,count in self.operators.coverage.items():
|
||||
stage,kind=name.split(':',1);self.coverage.setdefault(kind,{})[stage]=count
|
||||
self.timings['selected_cost']=final.total
|
||||
self.winner_evaluation=final.evaluation
|
||||
return tuple(phrases)
|
||||
|
||||
def phrase(self,states,index):
|
||||
actions=tuple(s.action for s in states);ns=tuple(n for a in actions for n in a.notes)
|
||||
start,end=actions[0].start,actions[-1].end
|
||||
last=ns[-1] if ns else None
|
||||
held=last and last.end-last.start>=2
|
||||
ending='handoff' if end==self.end else 'settle' if held else 'open'
|
||||
anchor=None
|
||||
if held:
|
||||
region=self.harmony.at(last.start)
|
||||
anchor=types.Anchor(f'gesture_{index}:arrival',last.start,last.start,last.pitch,
|
||||
frozenset({last.pitch.number()%12}),self.harmony.profile.register,last.end-last.start,
|
||||
'terminal' if ending=='handoff' else 'intermediate',region.chord,False,last.end,ending)
|
||||
evaluation=self.selector.complete(actions,states[-1].home,start=start,end=end)
|
||||
direction='up' if len(ns)>1 and ns[-1].pitch.number()>ns[0].pitch.number() else 'down'
|
||||
rhythm=tuple(g for a in actions for g in groups.describe(tuple((n.start-a.start,n.end-a.start) for n in a.notes),origin=a.start,source=a.kind))
|
||||
return types.Phrase(f'gesture_{index}',start,end,ns,actions[0].kind,direction,ending,anchor,
|
||||
tuple(c for a in actions for c in a.contracts),rhythm,'home',evaluation.total,evaluation,tuple(a.source or a.kind for a in actions))
|
||||
for phrase,introduced in self.candidates(start,end,state):
|
||||
total = state.total+phrase.evaluation.total*(end-start)
|
||||
successors.append(State((*state.phrases,phrase),state.home or (introduced if start>=0 and len(introduced)>=3 else ()),total))
|
||||
if not successors:
|
||||
raise SearchExhausted(f'No valid complete phrase at {start}–{end}; failures: {dict(self.failures)}')
|
||||
successors.sort(key=self.rank)
|
||||
states = tuple(generator.solo.sequences.diverse(successors,self.rank,
|
||||
lambda s:(s.phrases[-1].purpose,s.phrases[-1].direction,s.phrases[-1].ending),self.rules['histories']))
|
||||
for state in states:
|
||||
self.coverage[state.phrases[-1].purpose]['retained_histories'] += 1
|
||||
self.frontiers.append({'start':str(start),'end':str(end),'completed':len(successors),'retained':len(states),'best':str(min(s.total for s in states))})
|
||||
start = end
|
||||
self.completed = tuple(sorted(states,key=self.rank))
|
||||
winner = self.completed[0]
|
||||
self.timings['phrase_search_seconds'] = time.monotonic()-began
|
||||
self.winner_evaluation = generator.solo.selection.evaluate(winner.notes,self.start,self.end,self.harmony,
|
||||
home=winner.home,contracts=tuple(c for p in winner.phrases for c in p.expressions),center=self.phrases.center)
|
||||
for k,v in self.phrases.proposals.items():
|
||||
self.coverage[k]['proposed'] = v
|
||||
for p in winner.phrases:
|
||||
self.coverage[p.purpose]['selected'] += 1
|
||||
self.decisions = [{'start':str(p.start),'end':str(p.end),'purpose':p.purpose,'ending':p.ending,'evaluation':dataclasses.asdict(p.evaluation)} for p in winner.phrases]
|
||||
return winner.phrases
|
||||
|
||||
@@ -1,326 +0,0 @@
|
||||
"""Shared, enumerable musical actions for composition and derivation checking.
|
||||
|
||||
An action owns its complete rhythm and path. A rest is an action; an ordinary
|
||||
attack is not an implied cadence. Compound actions retain approach targets and
|
||||
whole tuplet groups. This module has no training/reference-file interface.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import dataclasses
|
||||
import functools
|
||||
from fractions import Fraction as F
|
||||
from collections import Counter
|
||||
|
||||
from generator import harmony as chords, vocabulary
|
||||
from generator.solo import types, expressions, groups, rhythm
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Action:
|
||||
start: F
|
||||
end: F
|
||||
notes: tuple
|
||||
kind: str
|
||||
contracts: tuple = ()
|
||||
source: str = ''
|
||||
|
||||
@functools.cached_property
|
||||
def key(self):
|
||||
return (self.start,self.end,self.kind,tuple((n.start,n.end,n.pitch.number()) for n in self.notes),
|
||||
tuple((c.rule,str(c.boundary),c.target.start,c.target.end,c.target.pitch.number()) for c in self.contracts))
|
||||
|
||||
|
||||
class Operators:
|
||||
def __init__(self,harmony,analysis,end,start=F(),enabled=True,sequences=None,rhythms=None):
|
||||
self.harmony=harmony;self.analysis=analysis;self.start=start;self.end=end;self.enabled=enabled
|
||||
self.rules=vocabulary.data('solo')['operators'];self.cache={};self.pitch_cache={};self.coverage=Counter()
|
||||
self.ordinary_cache={};self.validity_cache={}
|
||||
self.sequences=sequences
|
||||
self.licensed=frozenset(s.steps for values in sequences.paths.values() for s in values) if sequences else None
|
||||
self.patterns=[]
|
||||
if rhythms is not None:
|
||||
library=rhythm.Rhythms(rhythms,analysis.tempo)
|
||||
# Purposeful curated groups are peers; exhaustive catalogs contribute
|
||||
# representatives, not a probability proportional to XML file counts.
|
||||
candidates=library.groups
|
||||
for meter,values in library.catalog.items():
|
||||
if any(b-a==F(meter.split('/')[0])*4/int(meter.split('/')[1]) for a,b in zip(analysis.boundaries,analysis.boundaries[1:])):
|
||||
candidates+=list(values)
|
||||
seen=set()
|
||||
for name,pattern,length,_ in candidates:
|
||||
if 2<=len(pattern)<=4 and length<=4 and (pattern,length) not in seen:
|
||||
seen.add((pattern,length));self.patterns.append((name,pattern,length))
|
||||
|
||||
def pitches(self,time):
|
||||
span=self.harmony.at(time)
|
||||
if not span or not span.scale:return ()
|
||||
if span.scale not in self.pitch_cache:
|
||||
self.pitch_cache[span.scale]=tuple(p for i in range(-28,29)
|
||||
if self.harmony.profile.permits((p:=types.scale_pitch(span.scale,i)).number()))
|
||||
return self.pitch_cache[span.scale]
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def note(self,start,end,pitch,role):
|
||||
# Identity follows exact musical content, not an output filename or hash.
|
||||
return types.Note(start,end,pitch,role=role,event_id=f'{start}:{end}:{pitch.number()}')
|
||||
|
||||
def degree_pitches(self,chord,rule):
|
||||
spelling=expressions.degree_pitch(chord,rule['degree'],rule['alter'],60)
|
||||
# Enumerate octaves, not repeated rounding from C-octave centers. Halfway
|
||||
# pitches (e.g. F#) otherwise disappear because round uses even ties.
|
||||
return tuple(p for octave in range(1,6)
|
||||
if self.harmony.profile.permits((p:=dataclasses.replace(spelling,octave=octave)).number()))
|
||||
|
||||
def _accepted(self,action,validate=True):
|
||||
self.coverage['proposed:'+action.kind]+=1
|
||||
if action.end>self.end or action.end<=action.start:
|
||||
self.coverage['rejected_by_allocation:'+action.kind]+=1;return False
|
||||
if any(not self.harmony.profile.permits(n.pitch.number()) for n in action.notes):
|
||||
self.coverage['rejected_by_physical_domain:'+action.kind]+=1;return False
|
||||
if not groups.valid(tuple((n.start-action.start,n.end-action.start) for n in action.notes)):
|
||||
self.coverage['rejected_by_rhythm:'+action.kind]+=1;return False
|
||||
self.coverage['feasible:'+action.kind]+=1
|
||||
return self.is_valid(action) if validate else True
|
||||
|
||||
def is_valid(self,action):
|
||||
key=action.key
|
||||
if key in self.validity_cache:return self.validity_cache[key]
|
||||
if action.contracts:
|
||||
valid=self.harmony.evaluate(action.notes,expressions=action.contracts,complete=False,until=action.end).valid
|
||||
else:
|
||||
valid=True
|
||||
for n in action.notes:
|
||||
note_key=(n.start,n.end,n.pitch.number())
|
||||
if note_key not in self.ordinary_cache:self.ordinary_cache[note_key]=self.harmony.evaluate((n,),complete=True,until=n.end).valid
|
||||
if not self.ordinary_cache[note_key]:valid=False;break
|
||||
self.coverage[('valid:' if valid else 'rejected_by_rule:')+action.kind]+=1
|
||||
self.validity_cache[key]=valid
|
||||
if len(self.validity_cache)>8192:self.validity_cache.pop(next(iter(self.validity_cache)))
|
||||
return valid
|
||||
|
||||
@functools.lru_cache(maxsize=8192)
|
||||
def _color(self,n,span):
|
||||
result=[]
|
||||
for rule in vocabulary.data('harmony')['colors']:
|
||||
if rule.get('resolve') or not expressions.eligible_color(rule,span,self.harmony):continue
|
||||
expected=expressions.degree_pitch(expressions.parsed_chord(span.chord),rule['degree'],rule['alter'],n.pitch.number())
|
||||
if expected.number()!=n.pitch.number():continue
|
||||
c=types.ExpressionContract(n.event_id+':'+rule['id'],'color',rule['id'],(n,),n,span.chord)
|
||||
result.append(Action(n.start,n.end,(n,),'color',(c,),rule['id']))
|
||||
return tuple(result)
|
||||
|
||||
def proposals(self,time):
|
||||
return self.at(time,validate=False)
|
||||
|
||||
def at(self,time,eligible=None,first=None,*,validate=True):
|
||||
"""Unpruned legal actions. Incoming-state ranking is a separate operation."""
|
||||
if eligible is None and first is None and (time,validate) in self.cache:return self.cache[(time,validate)]
|
||||
span=self.harmony.at(time);values=[]
|
||||
if not span or not span.scale:
|
||||
end=min(self.end,span.end if span else self.end)
|
||||
result=(Action(time,end,(),'space'),)
|
||||
if eligible is not None:return tuple(a for a in result if eligible(a))
|
||||
self.cache[(time,validate)]=result;return result
|
||||
def add(action):
|
||||
if eligible is not None and not eligible(action):return
|
||||
if self._accepted(action,validate=validate):values.append(action)
|
||||
def entry(a,b,pitch):
|
||||
return first is None or first==(a,b,pitch.number())
|
||||
durations=tuple(map(F,self.rules['durations']))
|
||||
if self.analysis.tempo>vocabulary.data('roles')['difficulty']['sixteenth_max_bpm']:
|
||||
durations=tuple(d for d in durations if d>=F(1,2))
|
||||
for duration in durations:
|
||||
if time+duration>self.end:continue
|
||||
if (time*(4 if duration==F(1,4) else 2)).denominator==1:
|
||||
add(Action(time,time+duration,(),'rest'))
|
||||
for pitch in self.pitches(time):
|
||||
if not entry(time,time+duration,pitch):continue
|
||||
n=self.note(time,time+duration,pitch,'motion')
|
||||
if pitch.number()%12 in self.harmony.permitted(time,n.end):
|
||||
add(Action(time,n.end,(n,),'tone'))
|
||||
if self.enabled:
|
||||
delayed=vocabulary.data('lead_ins').get('delayed_approach')
|
||||
if delayed:
|
||||
for delay in map(F,delayed['arrivals']):
|
||||
arrival=time+delay
|
||||
for hold in map(F,delayed['holds']):
|
||||
if arrival+hold>self.end:continue
|
||||
region=self.harmony.at(arrival)
|
||||
if not region:continue
|
||||
for p in self.pitches(arrival):
|
||||
if p.number()%12 not in self.harmony.members(arrival,arrival+hold):continue
|
||||
target=self.note(arrival,arrival+hold,p,'arrival')
|
||||
for step in delayed['steps']:
|
||||
n=self.note(time,time+F(delayed['max_note']),expressions.chromatic(p,-step),'delayed_approach')
|
||||
if not entry(n.start,n.end,n.pitch):continue
|
||||
if n.pitch.number()%12 in self.harmony.permitted(n.start,n.end):continue
|
||||
c=types.ExpressionContract(n.event_id+':delayed','delayed_approach',delayed['id'],(n,),target,region.chord)
|
||||
add(Action(time,n.end,(n,),'delayed_approach',(c,),delayed['id']))
|
||||
# Optional colors retain functional degree identity independently of
|
||||
# the reference/exporter's enharmonic notation. No scale is widened.
|
||||
for rule in vocabulary.data('harmony')['colors']:
|
||||
if not expressions.eligible_color(rule,span,self.harmony):continue
|
||||
for p in self.degree_pitches(expressions.parsed_chord(span.chord),rule):
|
||||
if not self.harmony.profile.permits(p.number()):continue
|
||||
for duration in durations:
|
||||
if duration>F(rule['max_beats']) or time+duration>self.end:continue
|
||||
if not entry(time,time+duration,p):continue
|
||||
n=self.note(time,time+duration,p,'color')
|
||||
if not rule.get('resolve'):
|
||||
c=types.ExpressionContract(n.event_id+':'+rule['id'],'color',rule['id'],(n,),n,span.chord)
|
||||
add(Action(time,n.end,(n,),'color',(c,),rule['id']))
|
||||
else:
|
||||
for gap in (F(),F(1,2),F(1)):
|
||||
for q in self.pitches(n.end+gap):
|
||||
if not 0<abs(q.number()-p.number())<=2:continue
|
||||
for hold in map(F,self.rules['approach_holds']):
|
||||
target=self.note(n.end+gap,n.end+gap+hold,q,'arrival')
|
||||
target_span=self.harmony.at(target.start)
|
||||
if not target_span or target.end>self.end:continue
|
||||
c=types.ExpressionContract(n.event_id+':'+rule['id'],'color',rule['id'],(n,),target,target_span.chord)
|
||||
add(Action(time,n.end,(n,),'color_resolution',(c,),rule['id']))
|
||||
for boundary in (s.start for s in self.harmony.spans if time<s.start<=time+F(vocabulary.data('harmony')['anticipation_max'])):
|
||||
region=self.harmony.at(boundary)
|
||||
for hold in durations:
|
||||
for p in self.pitches(boundary):
|
||||
n=self.note(time,boundary,p,'anticipation')
|
||||
if not entry(n.start,n.end,p):continue
|
||||
target=self.note(boundary,boundary+hold,p,'arrival')
|
||||
c=types.ExpressionContract(n.event_id+':anticipation','anticipation','early_reattack',(n,),target,region.chord,boundary=boundary)
|
||||
add(Action(time,target.end,(n,target),'anticipation',(c,),c.rule))
|
||||
for duration in durations:
|
||||
if time+duration<=boundary:continue
|
||||
for p in self.pitches(boundary):
|
||||
if not entry(time,time+duration,p):continue
|
||||
n=self.note(time,time+duration,p,'anticipation')
|
||||
c=types.ExpressionContract(n.event_id+':anticipation','anticipation','early_chord_tone',(n,),n,region.chord,boundary=boundary)
|
||||
add(Action(time,n.end,(n,),'anticipation',(c,),c.rule))
|
||||
for rule in vocabulary.data('harmony')['colors']:
|
||||
if not expressions.eligible_color(rule,region,self.harmony):continue
|
||||
for p in self.degree_pitches(expressions.parsed_chord(region.chord),rule):
|
||||
if not entry(time,time+duration,p):continue
|
||||
n=self.note(time,time+duration,p,'anticipated_color')
|
||||
if not rule.get('resolve'):
|
||||
c=types.ExpressionContract(n.event_id+':anticipated_color','anticipated_color',rule['id'],(n,),n,region.chord,boundary=boundary)
|
||||
add(Action(time,n.end,(n,),'anticipated_color',(c,),rule['id']))
|
||||
else:
|
||||
for q in self.pitches(n.end):
|
||||
if not 0<abs(q.number()-p.number())<=2:continue
|
||||
for hold in map(F,self.rules['approach_holds']):
|
||||
target=self.note(n.end,n.end+hold,q,'arrival')
|
||||
c=types.ExpressionContract(n.event_id+':anticipated_color','anticipated_color',rule['id'],(n,),target,region.chord,boundary=boundary)
|
||||
add(Action(time,target.end,(n,target),'anticipated_color',(c,),rule['id']))
|
||||
for rule in vocabulary.data('lead_ins')['patterns']:
|
||||
if rule.get('difficulty') and self.analysis.tempo>vocabulary.data('roles')['difficulty'][rule['difficulty']+'_max_bpm']:continue
|
||||
# A connection's preceding event is explicit in the action. It cannot
|
||||
# obtain a fictitious preparation from a label or from a future note.
|
||||
if rule['kind']=='connection':
|
||||
for preparation in map(F,self.rules['connection_preparations']):
|
||||
if first is not None and (first[0]!=time or first[1]!=time+preparation):continue
|
||||
arrival=time+preparation+sum(map(F,rule['durations']),F())+F(rule['gap'])
|
||||
for hold in map(F,self.rules['approach_holds']):
|
||||
if arrival+hold>self.end:continue
|
||||
for p in self.pitches(arrival):
|
||||
region=self.harmony.at(arrival)
|
||||
target=self.note(arrival,arrival+hold,p,'arrival')
|
||||
cursor=time+preparation;notes=[]
|
||||
for step,length in zip(rule['semitones'],map(F,rule['durations'])):
|
||||
notes.append(self.note(cursor,cursor+length,expressions.chromatic(p,step),'connection'));cursor+=length
|
||||
for prep_pitch in self.pitches(time):
|
||||
if not entry(time,time+preparation,prep_pitch) or not 0<abs(prep_pitch.number()-notes[0].pitch.number())<=2:continue
|
||||
prep=self.note(time,time+preparation,prep_pitch,'preparation')
|
||||
c=types.ExpressionContract(prep.event_id+':'+rule['id'],'connection',rule['id'],tuple(notes),target,region.chord,preparation=prep)
|
||||
add(Action(time,target.end,(prep,*notes,target),'connection',(c,),rule['id']))
|
||||
continue
|
||||
gaps=tuple(map(F,rule.get('internal_gaps',[])))
|
||||
arrival=time+sum(map(F,rule['durations']),F())+sum(gaps,F())+F(rule['gap'])
|
||||
for hold in map(F,self.rules['approach_holds']):
|
||||
for p in self.pitches(arrival):
|
||||
if not entry(time,time+F(rule['durations'][0]),expressions.chromatic(p,rule['semitones'][0])):continue
|
||||
region=self.harmony.at(arrival)
|
||||
if not region:continue
|
||||
target=self.note(arrival,arrival+hold,p,'arrival')
|
||||
cursor=time;notes=[]
|
||||
for i,(step,length) in enumerate(zip(rule['semitones'],map(F,rule['durations']))):
|
||||
notes.append(self.note(cursor,cursor+length,expressions.chromatic(p,step),rule['kind']));cursor+=length+(gaps[i] if i<len(gaps) else 0)
|
||||
c=types.ExpressionContract(notes[0].event_id+':'+rule['id'],rule['kind'],rule['id'],tuple(notes),target,region.chord)
|
||||
add(Action(time,target.end,(*notes,target),rule['kind'],(c,),rule['id']))
|
||||
for colored in self._color(target,region):
|
||||
add(Action(time,target.end,(*notes,target),rule['kind'],(c,*colored.contracts),rule['id']))
|
||||
for upcoming in self.harmony.spans:
|
||||
if not arrival<upcoming.start<target.end or upcoming.start-arrival>F(vocabulary.data('harmony')['anticipation_max']):continue
|
||||
anticipation=types.ExpressionContract(target.event_id+':early','anticipation','early_chord_tone',(target,),target,upcoming.chord,boundary=upcoming.start)
|
||||
approach=dataclasses.replace(c,chord=upcoming.chord,boundary=upcoming.start)
|
||||
add(Action(time,target.end,(*notes,target),rule['kind'],(approach,anticipation),rule['id']))
|
||||
# Complete directional groups use exact scale indexing at each harmonic
|
||||
# time. A changed collection retains actual pitches, never degree labels.
|
||||
lengths_list=[(F(1,2),)*2,(F(1,2),)*4,(F(1),F(1,2),F(1,2)),(F(1,2),F(1),F(1,2))]
|
||||
difficulty=vocabulary.data('roles')['difficulty']
|
||||
if self.analysis.tempo<=difficulty['triplet_max_bpm']:lengths_list.append((F(1,3),)*3)
|
||||
if self.analysis.tempo<=difficulty['sixteenth_max_bpm']:lengths_list.extend(((F(1,4),)*4,(F(1,4),)*2))
|
||||
patterns=list(self.patterns)
|
||||
for lengths in lengths_list:
|
||||
cursor=F();pattern=[]
|
||||
for d in lengths:pattern.append((cursor,cursor+d));cursor+=d
|
||||
patterns.append(('complete_subdivision',tuple(pattern),cursor))
|
||||
for name,pattern,length in patterns:
|
||||
if time+length>self.end:continue
|
||||
if first is not None and (time+pattern[0][0],time+pattern[0][1])!=first[:2]:continue
|
||||
for stride in (-2,-1,1,2):
|
||||
# Catalog licenses relative paths, but never sets family probability.
|
||||
if self.licensed is not None and (stride,)*(len(pattern)-1) not in self.licensed:continue
|
||||
for initial in self.pitches(time):
|
||||
if not entry(time+pattern[0][0],time+pattern[0][1],initial):continue
|
||||
cursor=time;p=initial;notes=[]
|
||||
for index,(a,b) in enumerate(pattern):
|
||||
cursor=time+a
|
||||
region=self.harmony.at(cursor)
|
||||
if not region or not region.scale:break
|
||||
if index:
|
||||
try:p=types.scale_pitch(region.scale,types.scale_index(region.scale,p.number())+stride)
|
||||
except ValueError:break
|
||||
notes.append(self.note(cursor,time+b,p,'scalar' if abs(stride)==1 else 'scale_thirds'))
|
||||
if len(notes)==len(pattern):add(Action(time,time+length,tuple(notes),'scalar' if abs(stride)==1 else 'scale_thirds',source=f'{name}:degree_stride:{stride}'))
|
||||
# A real arpeggio walks successive members of the WRITTEN chord, not
|
||||
# alternate degrees of its selected seven-note collection.
|
||||
for direction in (-1,1):
|
||||
for initial in self.pitches(time):
|
||||
if initial.number()%12 not in span.roles.basic:continue
|
||||
if not entry(time+pattern[0][0],time+pattern[0][1],initial):continue
|
||||
p=initial;notes=[]
|
||||
for index,(a,b) in enumerate(pattern):
|
||||
region=self.harmony.at(time+a)
|
||||
if not region or not region.scale:break
|
||||
if index:
|
||||
choices=[q for q in self.pitches(time+a) if q.number()%12 in region.roles.basic and (q.number()-p.number())*direction>0]
|
||||
if not choices:break
|
||||
p=min(choices,key=lambda q:abs(q.number()-p.number()))
|
||||
notes.append(self.note(time+a,time+b,p,'arpeggio'))
|
||||
if len(notes)==len(pattern):add(Action(time,time+length,tuple(notes),'arpeggio',source=f'{name}:written_chord:{direction}'))
|
||||
# Equivalent musical actions need one stable derivation, not extra votes.
|
||||
unique={}
|
||||
for a in values:unique.setdefault(a.key,a)
|
||||
result=tuple(unique[k] for k in sorted(unique))
|
||||
if eligible is not None or first is not None:return result
|
||||
if len(self.cache)>=32:self.cache.pop(next(iter(self.cache)))
|
||||
if len(self.ordinary_cache)>8192:self.ordinary_cache.clear()
|
||||
self.cache[(time,validate)]=result
|
||||
return result
|
||||
|
||||
@functools.lru_cache(maxsize=512)
|
||||
def related(self,time,home,kind):
|
||||
"""Construct a heard-theme return or answer as a whole group, never a prefix."""
|
||||
if len(home)<3:return ()
|
||||
pattern=tuple((n.start-home[0].start,n.end-home[0].start) for n in home)
|
||||
if not groups.valid(pattern):return ()
|
||||
values=[]
|
||||
for p in self.pitches(time):
|
||||
notes=[]
|
||||
for n,(a,b) in zip(home,pattern):
|
||||
number=p.number()+(n.pitch.number()-home[0].pitch.number())*(1 if kind=='return' else -1)
|
||||
q=next((q for q in self.pitches(time+a) if q.number()==number),None)
|
||||
if q is None:break
|
||||
notes.append(self.note(time+a,time+b,q,kind))
|
||||
if len(notes)==len(home):
|
||||
action=Action(time,time+pattern[-1][1],tuple(notes),kind,source='realized_theme')
|
||||
if self._accepted(action):values.append(action)
|
||||
return tuple(values)
|
||||
@@ -0,0 +1,383 @@
|
||||
"""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.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])
|
||||
|
||||
|
||||
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.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 = (any(a[1]<b[0] for a,b in zip(pattern,pattern[1:])),any(a%1 for a,b in pattern),len({b-a for a,b in pattern})>1)
|
||||
representatives.setdefault(descriptor,(name,pattern,length,purposes))
|
||||
self.patterns += tuple(representatives.values())
|
||||
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=16384)
|
||||
def options(self,start,end,near,structural=False):
|
||||
result = []
|
||||
span = self.harmony.at(start)
|
||||
if not span or not span.scale:
|
||||
return ()
|
||||
for p in self.pitches(start,end):
|
||||
if abs(p.number()-near)>12:
|
||||
continue
|
||||
note = self.note(start,end,p,'arrival' if structural else 'motion')
|
||||
if self.harmony.evaluate((note,),complete=True).valid:
|
||||
if not structural or p.number()%12 in self.harmony.members(start,end):
|
||||
result.append((note,()))
|
||||
if self.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
|
||||
pitch = generator.solo.expressions.degree_pitch(generator.solo.expressions.parsed_chord(span.chord),rule['degree'],rule['alter'],p.number())
|
||||
if pitch.number()!=p.number():
|
||||
continue
|
||||
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((dataclasses.replace(note,role='color'),(color,)))
|
||||
return tuple(result)
|
||||
|
||||
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):
|
||||
"""DP compares completed paths; no single route step is committed alone."""
|
||||
if not rhythm:
|
||||
return None
|
||||
initial = incoming.pitch.number() if incoming else center
|
||||
if incoming and mode in ('connect','arpeggio'):
|
||||
direction = 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()))[:3]
|
||||
states = [(0,(n,),c,n.pitch.number()) for n,c in starts]
|
||||
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)]
|
||||
if degree_path:
|
||||
span = self.harmony.at(a)
|
||||
target = self.scale_step(span.scale,notes[-1].pitch.number(),expected) if span and span.scale else None
|
||||
if target is None:
|
||||
continue
|
||||
wanted = target.number()
|
||||
expected = wanted-notes[-1].pitch.number()
|
||||
for note,support in self.options(a,b,wanted):
|
||||
if degree_path and note.pitch.number()!=wanted:
|
||||
continue
|
||||
delta = note.pitch.number()-notes[-1].pitch.number()
|
||||
if mode=='arpeggio' and (note.pitch.number()%12 not in self.harmony.at(a).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 self.harmony.at(a).roles.basic and not support:
|
||||
error += 2
|
||||
next_states.append((cost+error,(*notes,note),(*contracts,*support),origin))
|
||||
if not next_states:
|
||||
return None
|
||||
states = sorted(next_states,key=lambda s:(s[0],tuple(n.pitch.number() for n in s[1])))[:8]
|
||||
for _,notes,contracts,_ in sorted(states,key=lambda s:(s[0],tuple(n.pitch.number() for n in s[1]))):
|
||||
if self.harmony.evaluate(notes,expressions=contracts,complete=True).valid:
|
||||
return notes,contracts
|
||||
return None
|
||||
|
||||
def contour(self,count,direction,mode,motif,time):
|
||||
if mode in ('statement','return','develop','answer') and motif:
|
||||
values = motif.intervals
|
||||
sign = -1 if mode=='answer' else 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 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 incoming:
|
||||
center = max(incoming.pitch.number()-7,min(incoming.pitch.number()+7,center))
|
||||
direction = self.choose((-1,1),'direction',(start,variant))
|
||||
hold,release = self.choose(self.arrivals,'arrival',(start,variant))
|
||||
close = end==self.end or self.choose((True,True,False),'closure',(start,variant))
|
||||
arrival = max(start,end-release-hold) if close and end-start>=4 else end
|
||||
notes,contracts,sources = [],[],[]
|
||||
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
|
||||
while cursor<arrival:
|
||||
span = self.harmony.at(cursor)
|
||||
if not span or not span.scale:
|
||||
cursor = min(arrival,span.end if span else arrival); continue
|
||||
remaining = arrival-cursor
|
||||
if group_index==0 and motif and motif.length<=remaining:
|
||||
pattern,length,source = motif.rhythm,motif.length,motif.source
|
||||
mode = purpose if purpose!='extend' else 'develop'
|
||||
else:
|
||||
def suitable(value):
|
||||
_,pattern,length,purposes = value
|
||||
fast = any((b-a).denominator not in (1,2) for a,b in pattern)
|
||||
return (length<=remaining and length>=min(remaining,1)
|
||||
and bool(set(purposes)&({'continue','connect','develop','answer','scalar','arpeggio'}|({'entrance'} if group_index==0 else set())))
|
||||
and (not fast or fast_groups<generator.vocabulary.data('solo')['subdivision_groups'])
|
||||
and generator.solo.groups.valid(tuple((cursor+a,cursor+b) for a,b in pattern)))
|
||||
eligible = [v for v in self.patterns if suitable(v)]
|
||||
if eligible:
|
||||
source,pattern,length,_ = self.choose(eligible,'rhythm',(start,variant,group_index))
|
||||
else:
|
||||
length = remaining; pattern,source = ((fractions.Fraction(),length),),'broadening'
|
||||
mode = self.choose(('connect','develop','arpeggio'),'movement',(start,variant,group_index))
|
||||
rhythm = tuple((cursor+a,cursor+b) for a,b in pattern)
|
||||
desired = self.contour(len(pattern),direction,mode,motif,cursor)
|
||||
phase = float((cursor-start)/max(1,end-start))
|
||||
local_center = center+direction*round(5*(1-abs(2*phase-1)))
|
||||
realized = self.route(rhythm,desired,notes[-1] if notes else incoming,local_center,mode,degree_path=mode=='connect')
|
||||
if realized is None:
|
||||
self.rejections['route'] += 1; return None
|
||||
ns,cs = realized
|
||||
if not signature_notes and len((*notes,*ns))>=3:
|
||||
signature_notes = tuple((*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)
|
||||
cursor += length; group_index += 1
|
||||
anchor = None; ending = 'open'
|
||||
if arrival<end:
|
||||
sounding_end = end-release
|
||||
previous = notes[-1].pitch.number() if notes else center
|
||||
options = self.options(arrival,sounding_end,previous,True)
|
||||
if not options:
|
||||
self.rejections['arrival'] += 1; return None
|
||||
def destination_cost(value):
|
||||
note,cs = value
|
||||
degrees = dict(self.harmony.at(arrival).roles.degrees).get(note.pitch.number()%12,())
|
||||
return (abs(note.pitch.number()-previous)+abs(note.pitch.number()-center)/3-int(bool(set(degrees)&{3,7})),note.pitch.number())
|
||||
target,support = sorted(options,key=destination_cost)[variant%min(3,len(options))]
|
||||
notes.append(target); contracts.extend(support)
|
||||
ending = 'handoff' if end==self.end else 'complete' if release else 'settle'
|
||||
anchor = generator.solo.types.Anchor('arrival:'+str(arrival),arrival,arrival,target.pitch,frozenset({target.pitch.number()%12}),
|
||||
self.harmony.profile.register,sounding_end-arrival,'terminal' if ending in ('complete','handoff') else 'intermediate',
|
||||
self.harmony.at(arrival).chord,False,sounding_end,ending)
|
||||
if not notes and any(s.scale and s.start<end and s.end>start for s in self.harmony.spans):
|
||||
return None
|
||||
if self.enabled and variant%3 and notes:
|
||||
notes,contracts = self.expression(tuple(notes),tuple(contracts),signature_notes,start,variant)
|
||||
checked = self.harmony.evaluate(notes,expressions=contracts,until=end,complete=True)
|
||||
if not checked.valid:
|
||||
self.rejections[checked.error] += 1; return None
|
||||
evaluation = generator.solo.selection.evaluate(notes,start,end,self.harmony,incoming=(incoming,) if incoming else (),
|
||||
home=home,purpose=purpose,contracts=contracts,center=self.center)
|
||||
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:
|
||||
groups = generator.solo.groups.describe(pattern,origin=start,source=purpose)
|
||||
except ValueError:
|
||||
self.rejections['rhythm_group'] += 1; return None
|
||||
phrase = generator.solo.types.Phrase(key,start,end,tuple(notes),purpose,'up' if direction>0 else 'down',ending,
|
||||
anchor,tuple(contracts),groups,'home' if home else key,float(evaluation.total),evaluation,tuple(sources))
|
||||
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
|
||||
@@ -4,7 +4,7 @@ import dataclasses, hashlib
|
||||
class Policy:
|
||||
long_note: float = 2.
|
||||
stopping_note: float = 1.
|
||||
composition_version: str = 'phrase-frontiers-2'
|
||||
composition_version: str = 'melodic-phrases-1'
|
||||
DEFAULT=Policy()
|
||||
def stable_rank(seed, value):
|
||||
return hashlib.sha256((seed + repr(value)).encode()).hexdigest()
|
||||
|
||||
@@ -25,24 +25,9 @@ class Instrument:
|
||||
def permits(self, number):
|
||||
return any(p.string_number in self.strings for p in generator.fretboard.single_positions_for_pitch(F(number)))
|
||||
|
||||
def cost(self, number):
|
||||
return .5 * min(1., max(0, abs(number - self.reference) - 7) / 12) if self.reference is not None else 0.
|
||||
|
||||
|
||||
UNRESTRICTED = Instrument()
|
||||
|
||||
|
||||
def weighted_center(notes):
|
||||
values = sorted((n.pitch.number(), n.end - n.start) for n in notes)
|
||||
total = sum((d for _, d in values), F())
|
||||
elapsed = F()
|
||||
for number, duration in values:
|
||||
elapsed += duration
|
||||
if elapsed >= total / 2:
|
||||
return number
|
||||
return 60
|
||||
|
||||
|
||||
def for_part(part, analysis=None):
|
||||
if part not in ('acoustic', 'electric'):
|
||||
raise ValueError('unknown solo instrument: ' + part)
|
||||
|
||||
@@ -36,7 +36,7 @@ def structure(solo):
|
||||
|
||||
def describe(solo,scales,measure_offset=0,time_offset=0,*,detail=None):
|
||||
detail=detail or structure(solo)
|
||||
lines=[f'Independent joint-action solo: {solo.part}', 'Times are exact quarter beats from the chorus start; negative times are allocated lead-ins. Pitches sound one octave below guitar notation.',
|
||||
lines=[f'Independent melodic-phrase solo: {solo.part}', 'Times are exact quarter beats from the chorus start; negative times are allocated lead-ins. Pitches sound one octave below guitar notation.',
|
||||
'Approximate fingering costs are not a biomechanical guarantee.',json.dumps(detail['phrasing'])]
|
||||
for p in solo.phrases:
|
||||
lines.append(f'{p.id}: {p.start}–{p.end}; {p.purpose}, {p.direction}, {p.ending}; theme {p.theme}; score {p.score:.4f}')
|
||||
|
||||
@@ -17,7 +17,7 @@ class Rhythms:
|
||||
self.groups.append((item['id'],tuple(pattern),cursor,tuple(item['purpose'])))
|
||||
for cell in vocabulary.rhythm_cells(tempo):
|
||||
# Imported authored cells remain whole. They are never sliced to fit.
|
||||
if groups.valid(cell.pattern):self.groups.append((cell.id,cell.pattern,cell.duration,('develop','answer') if cell.purpose=='answer' else ('arpeggio','scalar','develop','neighbor')))
|
||||
if groups.valid(cell.pattern):self.groups.append((cell.id,cell.pattern,cell.duration,(cell.purpose,)))
|
||||
# Meter-indexed catalog representatives: file counts do not set style frequency.
|
||||
self.catalog={}
|
||||
for meter,items in library.snippets.items():
|
||||
@@ -30,4 +30,3 @@ class Rhythms:
|
||||
descriptor=(len(pattern),pattern[0][0]>0,any(a[1]<b[0] for a,b in zip(pattern,pattern[1:])),len(set(b-a for a,b in pattern)))
|
||||
representatives.setdefault(descriptor,(item.id,pattern,bar.length,('develop','answer')))
|
||||
self.catalog[meter]=tuple(representatives.values())
|
||||
self.cache={}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""The authoritative action-evaluation record, independent of guitar travel."""
|
||||
"""The complete-phrase evaluation record, independent of guitar travel."""
|
||||
from dataclasses import dataclass
|
||||
from fractions import Fraction as F
|
||||
|
||||
|
||||
+43
-360
@@ -1,366 +1,49 @@
|
||||
"""Canonical sounded-event evaluation, shared by inference and training.
|
||||
"""One complete-event evaluator. No learned or constructor-family rewards."""
|
||||
import collections
|
||||
import fractions
|
||||
import generator.solo.scoring
|
||||
import generator.solo.types
|
||||
|
||||
Constructor names are not musical evidence. Complete equal-span candidates
|
||||
are compared using this record; local retrieval is explicitly only a heuristic.
|
||||
"""
|
||||
from collections import Counter
|
||||
from fractions import Fraction as F
|
||||
import dataclasses
|
||||
import functools
|
||||
from types import MappingProxyType
|
||||
from generator import vocabulary, harmony as chords, model
|
||||
from generator.solo import expressions, scoring, types
|
||||
|
||||
VERSION='phrase-selection-2'
|
||||
DEFAULT={'excessive_leap':12,'register_distance':1,'oscillation':6,'repeated_attack':3,
|
||||
'unprepared_space':12,'source_relation':-2,'theme_relation':-4,
|
||||
'sustained_arrival':-1,'directed_motion':-1}
|
||||
|
||||
CONDITIONS={
|
||||
'quality':frozenset(('major','minor','dominant','half_diminished','other')),
|
||||
'degree':frozenset(map(str,range(12))),
|
||||
'motion':frozenset(map(str,range(-12,13))),
|
||||
'metric':frozenset(('barline','beat','eighth','subdivision')),
|
||||
'duration':frozenset(str(F(i,4)).replace('/','_') for i in range(1,17))|{'1_3','2_3'},
|
||||
'source_pair':frozenset(('yes','no')),
|
||||
}
|
||||
|
||||
def rule_conditions(key):
|
||||
if not key.startswith('joint/'):return None
|
||||
pairs=tuple(p.split('=',1) for p in key[6:].split('&'))
|
||||
if not 2<=len(pairs)<=3 or any(len(p)!=2 for p in pairs):raise ValueError('Expected two or three contextual conditions')
|
||||
if len({p[0] for p in pairs})!=len(pairs) or list(pairs)!=sorted(pairs):raise ValueError('Noncanonical contextual rule')
|
||||
if any(k not in CONDITIONS or v not in CONDITIONS[k] for k,v in pairs):raise ValueError('Forbidden contextual rule condition')
|
||||
return dict(pairs)
|
||||
|
||||
|
||||
def validate_weights(weights):
|
||||
durations={str(F(i,4)).replace('/','_') for i in range(1,17)}|{'1_3','2_3'}
|
||||
allowed={f'duration/{d}' for d in durations}
|
||||
allowed|={f'harmony/{q}/{p}/{d}' for q in ('major','minor','dominant','half_diminished','other') for p in range(12) for d in durations}
|
||||
allowed|={f'motion/{p}/{d}' for p in range(-12,13) for d in durations}
|
||||
allowed|={f'metric/{m}/{d}' for m in ('barline','beat','eighth','subdivision') for d in durations}
|
||||
allowed|={f'register/{i}' for i in range(-4,5)}
|
||||
allowed|={f'obligation/step/{i}' for i in range(-12,13)}
|
||||
allowed|={f'obligation/delay/{str(F(i,4)).replace("/","_")}' for i in range(1,21)}
|
||||
allowed|={f'obligation/hold/{d}' for d in durations}
|
||||
allowed|={f'continuation/{a}/{b}' for a in ('up','down','repeat') for b in ('up','down','repeat')}
|
||||
allowed|={'source/rhythm_match','theme/realized_identity',*(f'space/{s}' for s in ('unharmonized','breath','after_arrival','delayed_entrance'))}
|
||||
for key,value in weights.items():
|
||||
joint=rule_conditions(key)
|
||||
if (key not in allowed and joint is None) or not isinstance(value,int) or isinstance(value,bool) or abs(value)>1024:
|
||||
raise ValueError('Unsupported fitted feature (only shared musical categories): '+key)
|
||||
if any(c not in 'abcdefghijklmnopqrstuvwxyz_0123456789/-=&' for c in key):
|
||||
raise ValueError('Noncanonical fitted feature: '+key)
|
||||
return dict(weights)
|
||||
VERSION = 'melodic-phrases-1'
|
||||
WEIGHTS = {'continuity': 2, 'repetition': 3, 'interruption': 2, 'activity': 2,
|
||||
'harmony': 2, 'register': 1, 'identity': 3}
|
||||
|
||||
|
||||
def signature(notes):
|
||||
if not notes:return ()
|
||||
first=notes[0]
|
||||
return tuple((n.start-first.start,n.end-n.start,n.pitch.number()-first.pitch.number()) for n in notes)
|
||||
notes = generator.solo.types.sounding_notes(notes)
|
||||
if not notes:
|
||||
return ()
|
||||
return tuple((n.start-notes[0].start, n.end-n.start, n.pitch.number()-notes[0].pitch.number()) for n in notes)
|
||||
|
||||
|
||||
def matches_signature(notes,expected):
|
||||
"""Exact signature equality, stopping at the first differing feature."""
|
||||
if len(notes)!=len(expected):return False
|
||||
if not notes:return True
|
||||
first=notes[0];pitch=first.pitch.number()
|
||||
return all(n.pitch.number()-pitch==relative and n.end-n.start==duration and n.start-first.start==onset
|
||||
for n,(onset,duration,relative) in zip(notes,expected))
|
||||
|
||||
|
||||
class Selector:
|
||||
@property
|
||||
def weights(self):return self._weights
|
||||
@weights.setter
|
||||
def weights(self,value):self._weights=MappingProxyType(validate_weights(value))
|
||||
|
||||
def __init__(self,harmony,analysis,center,weights=None):
|
||||
self.harmony=harmony;self.analysis=analysis;self.center=center
|
||||
self.weights=validate_weights(weights if weights is not None else vocabulary.role().get('selection_weights',{}))
|
||||
self.source_intervals=Counter(b.pitch.number()-a.pitch.number() for a,b in zip(analysis.melody,analysis.melody[1:]))
|
||||
self.source_pairs=Counter((b.pitch.number()-a.pitch.number(),c.pitch.number()-b.pitch.number())
|
||||
for a,b,c in zip(analysis.melody,analysis.melody[1:],analysis.melody[2:]))
|
||||
self.source_rhythms=Counter(n.end-n.start for n in analysis.melody)
|
||||
self.static={}
|
||||
self.home_signatures={}
|
||||
self.cost_static={};self.cost_weights=None
|
||||
|
||||
def contexts(self,notes,recent=()):
|
||||
history=list(recent[-2:])
|
||||
for n in notes:
|
||||
region=self.harmony.at(n.start)
|
||||
if region and region.chord:
|
||||
chord=expressions.parsed_chord(region.chord)
|
||||
root=int(model.NATURAL[chord['root']['letter']]+model.alteration(chord['root']['alter']))
|
||||
delta=n.pitch.number()-history[-1].pitch.number() if history else None
|
||||
previous=history[-1].pitch.number()-history[-2].pitch.number() if len(history)>1 else None
|
||||
metric='barline' if n.start+self.analysis.body_offset in self.analysis.boundaries else 'beat' if n.start%1==0 else 'eighth' if n.start%1==F(1,2) else 'subdivision'
|
||||
context={'quality':expressions.quality(chord),'degree':str((n.pitch.number()-root)%12),
|
||||
'metric':metric,'duration':str(n.end-n.start).replace('/','_'),
|
||||
'source_pair':'yes' if (previous,delta) in self.source_pairs else 'no'}
|
||||
if delta is not None:context['motion']=str(max(-12,min(12,delta)))
|
||||
yield context,int((n.end-n.start)*12)
|
||||
history.append(n)
|
||||
|
||||
def base(self,action):
|
||||
if action.key in self.static:return self.static[action.key]
|
||||
values=Counter();prior=Counter();ticks=int((action.end-action.start)*12)
|
||||
for contract in action.contracts:
|
||||
if contract.target.start>=action.end and contract.events:
|
||||
# Pending destinations must be visible to selection. Identical heard
|
||||
# attacks with different future obligations are different actions.
|
||||
last=contract.events[-1];target=contract.target
|
||||
values[f'obligation/step/{max(-12,min(12,target.pitch.number()-last.pitch.number()))}']+=ticks
|
||||
values['obligation/delay/'+str(target.start-last.start).replace('/','_')]+=ticks
|
||||
values['obligation/hold/'+str(target.end-target.start).replace('/','_')]+=ticks
|
||||
for n in action.notes:
|
||||
f,p=self.note_base(n.start,n.end,n.pitch.number())
|
||||
values.update(f);prior.update(p)
|
||||
self.static[action.key]=(values,prior)
|
||||
if len(self.static)>8192:self.static.pop(next(iter(self.static)))
|
||||
return values,prior
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def note_base(self,start,end,number):
|
||||
values=Counter();prior=Counter()
|
||||
duration=end-start;length=int(duration*12)
|
||||
label=str(duration).replace('/','_')
|
||||
values['duration/'+label]+=length
|
||||
beat=(start%1)
|
||||
metric='barline' if start+self.analysis.body_offset in self.analysis.boundaries else 'beat' if beat==0 else 'eighth' if beat==F(1,2) else 'subdivision'
|
||||
values['metric/'+metric+'/'+label]+=length
|
||||
band=max(-4,min(4,(number-self.center+6)//12))
|
||||
values[f'register/{band}']+=length
|
||||
region=self.harmony.at(start)
|
||||
if not region or not region.chord:return values,prior
|
||||
chord=expressions.parsed_chord(region.chord)
|
||||
root=int(model.NATURAL[chord['root']['letter']]+model.alteration(chord['root']['alter']))
|
||||
degree=(number-root)%12
|
||||
values[f'harmony/{expressions.quality(chord)}/{degree}/{label}']+=length
|
||||
prior['register_distance']+=length*max(0,abs(number-self.center)-7)
|
||||
if duration>=2 and number%12 in self.harmony.members(start,end):
|
||||
prior['sustained_arrival']+=length
|
||||
values['source/rhythm_match']+=length if duration in self.source_rhythms else 0
|
||||
return values,prior
|
||||
|
||||
def features(self,action,recent=(),home=(),previous_kind=''):
|
||||
if any(n.continuation for n in action.notes):action=dataclasses.replace(action,notes=types.sounding_notes(action.notes))
|
||||
static,base=self.base(action);values=Counter(static);prior=Counter(base)
|
||||
# Silence is measured as intervals in the sounded stream, independent of
|
||||
# the number of rest actions used to encode it. Missing harmony is distinct.
|
||||
cursor=action.start;before=recent[-1] if recent else None
|
||||
for n in (*action.notes,None):
|
||||
until=n.start if n else action.end
|
||||
if until>cursor:
|
||||
f,p=self.space_features(cursor,until,None if before is None else before.end-before.start>=F(3,2))
|
||||
values.update(f);prior.update(p)
|
||||
if n:cursor=n.end;before=n
|
||||
previous=recent[-1] if recent else None
|
||||
intervals=[b.pitch.number()-a.pitch.number() for a,b in zip(recent,recent[1:])]
|
||||
for n in action.notes:
|
||||
length=int((n.end-n.start)*12)
|
||||
if previous:
|
||||
delta=n.pitch.number()-previous.pitch.number()
|
||||
token=max(-12,min(12,delta));label=str(n.end-n.start).replace('/','_')
|
||||
values[f'motion/{token}/{label}']+=length
|
||||
if intervals:
|
||||
# Direction changes are contextual, not a blanket leap prohibition.
|
||||
sign=lambda x: 'up' if x>0 else 'down' if x<0 else 'repeat'
|
||||
values[f'continuation/{sign(intervals[-1])}/{sign(delta)}']+=length
|
||||
if len(intervals)>=2 and intervals[-2]==-intervals[-1]==delta and delta:prior['oscillation']+=length
|
||||
prior['excessive_leap']+=length*max(0,abs(delta)-7)
|
||||
prior['repeated_attack']+=length if delta==0 and n.start==previous.end else 0
|
||||
# Ordered relationships, not the set of available intervals. Weight
|
||||
# by sounding duration so subdivisions cannot multiply a reward.
|
||||
if intervals and (intervals[-1],delta) in self.source_pairs:prior['source_relation']+=length
|
||||
if 0<abs(delta)<=2 and intervals and delta*intervals[-1]>0:prior['directed_motion']+=length
|
||||
intervals.append(delta)
|
||||
previous=n
|
||||
if len(home)>=3:
|
||||
key=id(home)
|
||||
if key not in self.home_signatures:
|
||||
self.home_signatures[key]=(home,signature(home))
|
||||
if len(self.home_signatures)>32:self.home_signatures.pop(next(iter(self.home_signatures)))
|
||||
expected=self.home_signatures[key][1]
|
||||
joined=(*recent,*action.notes)
|
||||
for i in range(max(0,len(recent)-len(home)+1),len(joined)-len(home)+1):
|
||||
part=joined[i:i+len(home)]
|
||||
if part[0].start>home[-1].end and matches_signature(part,expected):
|
||||
values['theme/realized_identity']+=12
|
||||
prior['theme_relation']+=12
|
||||
joint=[(key,rule_conditions(key)) for key in self.weights if key.startswith('joint/')]
|
||||
if joint:
|
||||
for context,ticks in self.contexts(action.notes,recent):
|
||||
for key,conditions in joint:
|
||||
if all(context.get(k)==v for k,v in conditions.items()):values[key]+=ticks
|
||||
return values,prior
|
||||
|
||||
def complete(self,actions,home=(),*,start=None,end=None):
|
||||
"""The sole authoritative score. Ties/rests/partitions are canonicalized."""
|
||||
from generator.solo.operators import Action
|
||||
actions=tuple(actions)
|
||||
first=start if start is not None else actions[0].start
|
||||
last=end if end is not None else actions[-1].end
|
||||
notes=types.sounding_notes(tuple(n for a in actions for n in a.notes))
|
||||
contracts=tuple(dict.fromkeys(c for a in actions for c in a.contracts))
|
||||
return self.evaluate(Action(first,last,notes,'sounded_phrase',contracts),home=home)
|
||||
|
||||
def extend(self,previous,old_notes,action,home=(),*,old_home=(),start=F(),contracts=()):
|
||||
"""Exact sufficient-statistic update, checked against complete() in tests.
|
||||
|
||||
Reevaluate the seam and the tail containing any possible new home match.
|
||||
Previously completed events are not rescored on every search expansion.
|
||||
"""
|
||||
from generator.solo.operators import Action
|
||||
if previous is None or home!=old_home or any(n.continuation for n in action.notes):
|
||||
notes=types.sounding_notes((*old_notes,*action.notes))
|
||||
return self.evaluate(Action(start,action.end,notes,'sounded_phrase',contracts),home=home)
|
||||
span=F(previous.denominators['span'])
|
||||
raw=Counter({k:v*span for k,v in previous.raw.items() if not k.startswith('obligation/')})
|
||||
index=max(0,len(old_notes)-max(3,len(home)+2))
|
||||
tail=old_notes[index:];context=old_notes[max(0,index-3):index]
|
||||
begin=tail[0].start if index else start
|
||||
before=Action(begin,action.start,tail,'sounded_phrase')
|
||||
after=Action(begin,action.end,(*tail,*action.notes),'sounded_phrase')
|
||||
for sign,part in ((-1,before),(1,after)):
|
||||
f,p=self.features(part,context,home)
|
||||
for k,v in f.items():
|
||||
if not k.startswith('obligation/'):raw[k]+=sign*v
|
||||
for k,v in p.items():raw['prior/'+k]+=sign*v
|
||||
obligations,_=self.base(Action(start,action.end,(),'sounded_phrase',contracts))
|
||||
raw.update({k:v for k,v in obligations.items() if k.startswith('obligation/')})
|
||||
weights={'prior/'+k:v for k,v in DEFAULT.items()}|self.weights
|
||||
duration=action.end-start
|
||||
raw={k:F(v)/duration for k,v in raw.items() if v}
|
||||
contributions={k:v*weights.get(k,0) for k,v in raw.items()}
|
||||
return scoring.Evaluation(VERSION,True,raw,weights,contributions,sum(contributions.values(),F()),
|
||||
{'span':str(duration),'ticks_per_quarter':12,'attacks':len(old_notes)+len(action.notes)})
|
||||
|
||||
def evaluate(self,action,recent=(),home=(),previous_kind=''):
|
||||
features,prior=self.features(action,recent,home,previous_kind)
|
||||
contributions={'prior/'+k:v*DEFAULT[k] for k,v in prior.items() if v}
|
||||
contributions.update({k:v*self.weights.get(k,0) for k,v in features.items() if v})
|
||||
weights={'prior/'+k:v for k,v in DEFAULT.items()}|self.weights
|
||||
raw={'prior/'+k:v for k,v in prior.items() if v}|{k:v for k,v in features.items() if v}
|
||||
span=action.end-action.start
|
||||
contributions={k:F(v)/span for k,v in contributions.items()}
|
||||
raw={k:F(v)/span for k,v in raw.items()}
|
||||
total=sum(contributions.values(),F())
|
||||
return scoring.Evaluation(VERSION,True,raw,weights,contributions,total,
|
||||
{'span':str(action.end-action.start),'ticks_per_quarter':12,'attacks':len(types.sounding_notes(action.notes))})
|
||||
|
||||
def cost(self,action,recent=(),home=(),previous_kind=''):
|
||||
"""Identical selection total without serializing every rejected breakdown."""
|
||||
features,prior=self.features(action,recent,home,previous_kind)
|
||||
numerator=sum(v*DEFAULT[k] for k,v in prior.items())
|
||||
numerator+=sum(v*self.weights.get(k,0) for k,v in features.items())
|
||||
return F(numerator)/(action.end-action.start)
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def transition_cost(self,delta,preceding,earlier,length,label,contiguous):
|
||||
token=max(-12,min(12,delta))
|
||||
sign=lambda x:'up' if x>0 else 'down' if x<0 else 'repeat'
|
||||
cost=length*self.weights.get(f'motion/{token}/{label}',0)
|
||||
cost+=length*DEFAULT['excessive_leap']*max(0,abs(delta)-7)
|
||||
if delta==0 and contiguous:cost+=length*DEFAULT['repeated_attack']
|
||||
if preceding is not None:
|
||||
cost+=length*self.weights.get(f'continuation/{sign(preceding)}/{sign(delta)}',0)
|
||||
if (preceding,delta) in self.source_pairs:cost+=length*DEFAULT['source_relation']
|
||||
if 0<abs(delta)<=2 and delta*preceding>0:cost+=length*DEFAULT['directed_motion']
|
||||
if earlier is not None and earlier==-preceding==delta and delta:cost+=length*DEFAULT['oscillation']
|
||||
return cost
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def seam_cost(self,prefix,context,contiguous):
|
||||
def total(initial):
|
||||
pitches=list(initial);value=0
|
||||
for index,(number,length,label,touch) in enumerate(prefix):
|
||||
if pitches:
|
||||
value+=self.transition_cost(number-pitches[-1],pitches[-1]-pitches[-2] if len(pitches)>1 else None,
|
||||
pitches[-2]-pitches[-3] if len(pitches)>2 else None,length,label,contiguous if index==0 and initial else touch)
|
||||
pitches.append(number)
|
||||
return value
|
||||
return total(context)-total(())
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def unary_cost(self,start,end,number):
|
||||
features,prior=self.note_base(start,end,number)
|
||||
return sum(v*DEFAULT[k] for k,v in prior.items())+sum(v*self.weights.get(k,0) for k,v in features.items())
|
||||
|
||||
def intrinsic_cost(self,action):
|
||||
"""Exact ordinary retrieval cost without constructing discarded reports.
|
||||
|
||||
Unary, transition, silence and pending-target contributions are identical
|
||||
to features(action). Contextual learned conjunctions use the full path.
|
||||
"""
|
||||
notes=types.sounding_notes(action.notes) if any(n.continuation for n in action.notes) else action.notes
|
||||
cost=0;history=[];cursor=action.start;settled=None;previous=None
|
||||
for n in notes:
|
||||
if n.start>cursor:cost+=self.gap_cost(cursor,n.start,settled)
|
||||
number=n.pitch.number();length=int((n.end-n.start)*12);label=str(n.end-n.start).replace('/','_')
|
||||
cost+=self.unary_cost(n.start,n.end,number)
|
||||
if history:
|
||||
cost+=self.transition_cost(number-history[-1],history[-1]-history[-2] if len(history)>1 else None,
|
||||
history[-2]-history[-3] if len(history)>2 else None,length,label,n.start==previous.end)
|
||||
history.append(number);previous=n;cursor=n.end;settled=n.end-n.start>=F(3,2)
|
||||
if action.end>cursor:cost+=self.gap_cost(cursor,action.end,settled)
|
||||
ticks=int((action.end-action.start)*12)
|
||||
for contract in action.contracts:
|
||||
if contract.target.start>=action.end and contract.events:
|
||||
last=contract.events[-1];target=contract.target
|
||||
cost+=ticks*self.weights.get(f'obligation/step/{max(-12,min(12,target.pitch.number()-last.pitch.number()))}',0)
|
||||
cost+=ticks*self.weights.get('obligation/delay/'+str(target.start-last.start).replace('/','_'),0)
|
||||
cost+=ticks*self.weights.get('obligation/hold/'+str(target.end-target.start).replace('/','_'),0)
|
||||
return cost
|
||||
|
||||
def fast_cost(self,action,recent=(),context=None):
|
||||
"""Unnormalized endpoint-bucket retrieval, with cached internal routes.
|
||||
|
||||
Only the first three transitions depend on incoming history. Final ranking
|
||||
always uses complete(); this heuristic does not compare unequal spans.
|
||||
"""
|
||||
# Learned conjunctions include incoming context; do not use a shortcut
|
||||
# that forgets it. This small fallback is explicit and profileable.
|
||||
if any(n.continuation for n in action.notes) or any(k.startswith('joint/') for k in self.weights):
|
||||
f,p=self.features(action,recent)
|
||||
return sum(v*DEFAULT[k] for k,v in p.items())+sum(v*self.weights.get(k,0) for k,v in f.items())
|
||||
if self.cost_weights is not self.weights:
|
||||
self.cost_static.clear();self.transition_cost.cache_clear();self.seam_cost.cache_clear();self.gap_cost.cache_clear();self.unary_cost.cache_clear();self.cost_weights=self.weights
|
||||
key=id(action)
|
||||
if key not in self.cost_static:
|
||||
value=self.intrinsic_cost(action)
|
||||
prefix=tuple((n.pitch.number(),int((n.end-n.start)*12),str(n.end-n.start).replace('/','_'),
|
||||
i>0 and action.notes[i-1].end==n.start) for i,n in enumerate(action.notes[:3]))
|
||||
self.cost_static[key]=(action,value,prefix)
|
||||
if len(self.cost_static)>32768:self.cost_static.pop(next(iter(self.cost_static)))
|
||||
_,value,prefix=self.cost_static[key]
|
||||
if context is None:context=(tuple(n.pitch.number() for n in recent[-3:]),recent[-1].end if recent else None)
|
||||
if context[0] and prefix:value+=self.seam_cost(prefix,context[0],context[1]==action.notes[0].start)
|
||||
gap_end=action.notes[0].start if action.notes else action.end
|
||||
if recent and gap_end>action.start:
|
||||
# Incoming sustained arrivals change the role of a leading rest, just
|
||||
# as incoming pitches change the first intervals. Do not cache it as a
|
||||
# fresh silent entrance in every action.
|
||||
value+=self.gap_cost(action.start,gap_end,recent[-1].end-recent[-1].start>=F(3,2))-self.gap_cost(action.start,gap_end,None)
|
||||
return value
|
||||
|
||||
@functools.lru_cache(maxsize=8192)
|
||||
def space_features(self,start,end,settled):
|
||||
values=Counter();prior=Counter();delayed=0
|
||||
boundaries=sorted({start,end,*[s.start for s in self.harmony.spans if start<s.start<end],
|
||||
*[s.end for s in self.harmony.spans if start<s.end<end]})
|
||||
for left,right in zip(boundaries,boundaries[1:]):
|
||||
ticks=int((right-left)*12);region=self.harmony.at(left)
|
||||
label='unharmonized' if not region or not region.scale else 'delayed_entrance' if settled is None else 'after_arrival' if settled else 'breath'
|
||||
values['space/'+label]+=ticks
|
||||
if label=='breath':prior['unprepared_space']+=ticks
|
||||
if label=='delayed_entrance':delayed+=ticks
|
||||
prior['unprepared_space']+=max(0,delayed-12)
|
||||
return values,prior
|
||||
|
||||
@functools.lru_cache(maxsize=8192)
|
||||
def gap_cost(self,start,end,settled):
|
||||
f,p=self.space_features(start,end,settled)
|
||||
return sum(v*DEFAULT[k] for k,v in p.items())+sum(v*self.weights.get(k,0) for k,v in f.items())
|
||||
def evaluate(notes, start, end, harmony, *, incoming=(), home=(), purpose='develop', contracts=(), center=60):
|
||||
notes = generator.solo.types.sounding_notes(notes)
|
||||
duration = end-start
|
||||
if duration <= 0:
|
||||
raise ValueError('Evaluation requires a positive span')
|
||||
checked = harmony.evaluate(notes, expressions=contracts, until=end, complete=True)
|
||||
sounding = sum((n.end-n.start for n in notes), fractions.Fraction())
|
||||
pitches = [n.pitch.number() for n in notes]
|
||||
joined = [n.pitch.number() for n in incoming[-1:]]+pitches
|
||||
intervals = [b-a for a,b in zip(joined, joined[1:])]
|
||||
continuity = sum(max(0, abs(d)-7) for d in intervals)/max(1,len(intervals))
|
||||
oscillation = sum(a == -b == c and a != 0 for a,b,c in zip(intervals,intervals[1:],intervals[2:]))
|
||||
figures = collections.Counter(tuple(pitches[i:i+4]) for i in range(max(0,len(pitches)-3)))
|
||||
repetition = (oscillation+sum(a==b for a,b in zip(pitches,pitches[1:]))+sum(max(0,n-2) for n in figures.values()))/max(1,len(notes))
|
||||
interruption = sum(a.end<b.start and a.end-a.start<1 and b.end-b.start<1 for a,b in zip(notes,notes[1:]))/max(1,len(notes))
|
||||
density = fractions.Fraction(len(notes)*4,1)/duration
|
||||
active = any(s.scale and s.start<end and s.end>start for s in harmony.spans)
|
||||
activity = max(0,density-7)+(2 if active and not notes else 0)
|
||||
if notes and sounding<duration/3:
|
||||
activity += (duration/3-sounding)/duration
|
||||
identity = 0
|
||||
if home and purpose == 'return':
|
||||
sample = notes[:len(home)]
|
||||
identity = 2 if len(sample)!=len(home) else sum((a[0],a[1])!=(b[0],b[1]) or (a[2]>0)!=(b[2]>0) for a,b in zip(signature(home),signature(sample)))/len(home)
|
||||
register = sum(float(n.end-n.start)*max(0,abs(n.pitch.number()-center)-10) for n in notes)/max(1,float(sounding))
|
||||
values = dict(continuity=continuity,repetition=repetition,interruption=interruption,
|
||||
activity=activity,harmony=checked.exposure,register=register,identity=identity)
|
||||
raw = {k:fractions.Fraction(str(v)).limit_denominator(1000000) for k,v in values.items()}
|
||||
contributions = {k:v*WEIGHTS[k] for k,v in raw.items()}
|
||||
return generator.solo.scoring.Evaluation(VERSION,checked.valid,raw,dict(WEIGHTS),contributions,
|
||||
sum(contributions.values(),fractions.Fraction()),{'span':str(duration),'attacks':len(notes),'sounding':str(sounding)})
|
||||
|
||||
@@ -5,8 +5,6 @@ import pathlib
|
||||
import types
|
||||
import functools
|
||||
|
||||
import generator.solo.policy
|
||||
|
||||
import generator.model
|
||||
import generator.musicxml
|
||||
|
||||
@@ -24,30 +22,6 @@ class Sequence:
|
||||
offsets = self.offsets()
|
||||
return max(map(abs, self.steps)), max(offsets) - min(offsets), sum(a * b < 0 for a, b in zip(self.steps, self.steps[1:])), self.id
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Path:
|
||||
steps: tuple[int, ...]
|
||||
sources: tuple[str, ...]
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def offsets(self):
|
||||
result = [0]
|
||||
for step in self.steps:
|
||||
result.append(result[-1] + step)
|
||||
return tuple(result)
|
||||
|
||||
@functools.lru_cache(maxsize=32768)
|
||||
def family(steps):
|
||||
if not steps or all(s == 0 for s in steps):
|
||||
return 'repeated'
|
||||
turns = sum(a * b < 0 for a, b in zip(steps, steps[1:]))
|
||||
if not turns:
|
||||
return 'rising' if sum(steps) > 0 else 'falling'
|
||||
if max(map(abs, steps)) >= 4:
|
||||
return 'leap_return'
|
||||
if sum(steps) == 0 and max(map(abs, steps)) == 1:
|
||||
return 'neighbor'
|
||||
return ('arch' if steps[0] > 0 else 'valley') if turns == 1 else 'wave'
|
||||
|
||||
def diverse(items, rank, category, limit):
|
||||
"""Round-robin eligible musical families before a destructive cutoff."""
|
||||
groups = {}
|
||||
@@ -70,7 +44,6 @@ def diverse(items, rank, category, limit):
|
||||
@dataclasses.dataclass(frozen=True, init=False)
|
||||
class Library:
|
||||
paths: types.MappingProxyType
|
||||
chains: dict
|
||||
def __init__(self, sequences: tuple[Sequence, ...]):
|
||||
paths, identities = {}, set()
|
||||
for sequence in sequences:
|
||||
@@ -79,7 +52,6 @@ class Library:
|
||||
identities.add(sequence.id)
|
||||
paths.setdefault(len(sequence.steps) + 1, []).append(sequence)
|
||||
object.__setattr__(self, 'paths', types.MappingProxyType({count: tuple(sorted(items, key=lambda s: s.rank())) for count, items in paths.items()}))
|
||||
object.__setattr__(self, 'chains', {})
|
||||
def connecting(self, count: int, displacement: int | None = None) -> tuple[Sequence, ...]:
|
||||
return tuple(s for s in self.paths.get(count, ()) if displacement is None or s.offsets()[-1] == displacement)
|
||||
|
||||
|
||||
@@ -32,8 +32,7 @@ def _load(contents):
|
||||
cells=result['rhythms']['cells']+result['solo']['cells']
|
||||
if len({c['id'] for c in cells})!=len(cells):raise ValueError('Duplicate rhythm identity')
|
||||
rules=result['solo']
|
||||
if any(type(rules['operators'][k]) is not int or rules['operators'][k]<=0 for k in ('histories','completed','per_state')):raise ValueError('Invalid deterministic search budget')
|
||||
if any(F(d)<=0 for d in rules['operators']['durations']):raise ValueError('Invalid action duration')
|
||||
if any(type(rules['search'][k]) is not int or rules['search'][k]<=0 for k in ('histories','alternatives')):raise ValueError('Invalid deterministic search budget')
|
||||
for group in rules['groups']:
|
||||
if not group['values'] or any(F(v.removeprefix('r:'))<=0 for v in group['values']):raise ValueError('Invalid rhythm group')
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user