diff --git a/audition/fit_selection.py b/audition/fit_selection.py index 2578dcf..f500dcd 100644 --- a/audition/fit_selection.py +++ b/audition/fit_selection.py @@ -7,22 +7,20 @@ normal composer; accepting injected pitches is not a construction proof. from __future__ import annotations import argparse import bisect -from collections import Counter import dataclasses from fractions import Fraction as F import hashlib import heapq import json -import math from pathlib import Path import time from audition.reference_review import read -from audition.solo_compare import events, rests -from generator import accompaniment, model, musicxml, parts, production, resources, score, vocabulary +from audition.solo_compare import events +from generator import accompaniment, musicxml, parts, production, resources, score, vocabulary from generator.arrange import form, generate from generator.scales import selection, timeline -from generator.solo import analysis, construction, harmonic, profiles, report, types +from generator.solo import analysis, construction, harmonic, profiles def build(source_path,role,data_dir,cache_dir): @@ -54,122 +52,50 @@ def canonical(notes): def derive(builder,target): - """Search grammar derivations constrained to the exact supplied events. + """Stateful target-constrained coverage probe; NEVER production success. - This is a development diagnostic, not normal production inference. Failure - is recorded at its earliest uncovered event. No target action is injected. + Separate histories carry actual pending contracts. This bounded diagnostic + can report exhaustion, not prove a phrase impossible. No notes are injected. """ encoded=tuple((F(n['start']),F(n['end']),int(n['pitch'])) for n in events(target)) starts=tuple(n[0] for n in encoded);encoded_set=set(encoded) - queue=[builder.start];paths={builder.start:()};seen=set();furthest=builder.start + queue=[builder.start];chart={builder.start:[construction.State()]};queued={builder.start} + furthest=builder.start;last=construction.State();cuts=0 while queue: - at=heapq.heappop(queue) - if at in seen:continue - seen.add(at);furthest=max(furthest,at) + at=heapq.heappop(queue);states=chart.pop(at);furthest=max(furthest,at) + unique={builder.identity(s):s for s in states};states=list(unique.values()) + cuts+=max(0,len(states)-16);states=states[:16] if at==builder.end: - path=paths[at] - checked=builder.harmony.evaluate(tuple(n for a in path for n in a.notes),expressions=tuple(c for a in path for c in a.contracts),complete=True,until=at) - return path,{'complete':checked.valid,'covered_until':str(at),'actions':len(path), - 'status':'constructible' if checked.valid else 'rejected_by_full_context','error':checked.error} + for state in states: + checked=builder.harmony.evaluate(state.notes,expressions=tuple(c for a in state.actions for c in a.contracts),complete=True,until=at) + if checked.valid and canonical(state.notes)==encoded: + return state.actions,{'complete':True,'covered_until':str(at),'actions':len(state.actions), + 'status':'statefully_constructible','diagnostic_only':True,'pruned_histories':cuts} left=bisect.bisect_left(starts,at) def matches(action): if any((c.target.start,c.target.end,c.target.pitch.number()) not in encoded_set for c in action.contracts):return False expected=encoded[left:bisect.bisect_left(starts,action.end)] return not (left and encoded[left-1][1]>at) and canonical(action.notes)==expected first=encoded[left] if left=furthest),None) - reasons=[] - if next_event: - start,end,pitch=next_event;region=builder.harmony.at(start) - reasons=[{'pitch':pitch,'start':str(start),'end':str(end), - 'source_chord':region.chord if region else None,'mapped':bool(region and region.scale and pitch%12 in region.scale.pitches), - 'basic':bool(region and pitch%12 in region.roles.basic), - 'function':builder.harmony.function(pitch,region) if region else 'missing_harmony'}] - return paths.get(furthest,()),{'complete':False,'covered_until':str(furthest), - 'status':'not_proposed_or_rejected_by_rule','next_events':reasons, - 'budget':'Exhaustive derivation through available unpruned grammar actions, not unrestricted music search.', - 'action_coverage':dict(builder.operators.coverage)} + return last.actions,{'complete':False,'covered_until':str(furthest), + 'status':'coverage_search_exhausted' if cuts else 'not_proposed_or_rejected_by_rule', + 'next_event':tuple(map(str,next_event)) if next_event else None, + 'pruned_histories':cuts,'diagnostic_only':True,'action_coverage':dict(builder.operators.coverage)} -def fit(examples,passes,on_checkpoint=None): - """Integer early-update structured perceptron on shared musical features. - - Each roll-in is normal source-only ranking until its first wrong action. - Only the development command finds a compatible positive action and updates - coefficients. Production never receives the positive prefix. A zero-error - pass means every unrestricted roll-in produced the entire expected stream. - """ - weights={r:dict(b.selector.weights) for _,r,b,_ in examples};history=[];best=None;tables={} - # Whole actions include pickup+hold and compound groups (e.g. 17/4). - # Eight beats on the supported twelfth-beat lattice need numerators up to - # 96. The previous LCM 1..16 covered note values, not complete actions. - normalization=math.lcm(*range(1,97)) - def table(name,b,at,state): - key=(name,at,canonical(state.recent),canonical(state.home),state.last_kind,state.pending) - if key in tables:return tables[key] - rows=[] - actions=b.operators.at(at) - if state.home and state.last_kind!='return':actions+=b.operators.related(at,state.home,'return')+b.operators.related(at,state.home,'answer') - for action in actions: - if b.advance(state,None,action) is None:continue - features,prior=b.selector.features(action,state.recent,state.home,state.last_kind) - factor=F(normalization)/(action.end-action.start) - if factor.denominator!=1:raise ValueError('Unsupported training normalization span') - factor=int(factor) - rows.append((action,tuple((k,v*factor) for k,v in features.items() if v), - sum(v*b.selector_default[k] for k,v in prior.items())*factor)) - if len(tables)>=24:tables.pop(next(iter(tables))) - tables[key]=tuple(rows);return tables[key] - from generator.solo.selection import DEFAULT - for _,_,b,_ in examples:b.selector_default=DEFAULT - for iteration in range(passes): - updates={r:Counter() for r in weights};records=[];errors=0;covered=0 - for name,role,b,gold in examples: - b.selector.weights=weights[role];state=construction.State();at=b.start - encoded=canonical(tuple(n for a in gold for n in a.notes));encoded_set=set(encoded);starts=tuple(n[0] for n in encoded) - while at=builder.end-builder.start:break + selected.append(m);length+=model.fraction(m['duration']) + if length!=builder.end-builder.start:raise ValueError('Diagnostic export requires complete source measures') + records=[] + for index,state in enumerate(builder.completed,1): + played=placement.place(state.notes,builder.analysis.tempo,profile=builder.harmony.profile) + if played is None:raise ValueError('Completed candidate is unplayable') + shifted=tuple(dataclasses.replace(n,start=n.start-builder.start,end=n.end-builder.start) for n in played[1]) + part=copy.deepcopy(document);staff=part['staves'][0] + part['staves']=[staff];staff['measures']=notation.measures(selected,shifted,'candidate') + staff['tuplets']=notation.tuplets(staff['measures']);staff['tablature']=True + for key in ('scale_spans','octave_shifts','slurs','wedges'):staff.pop(key,None) + path=output/f'candidate-{index}.xml' + musicxml_writer.write_musicxml(path,part,tablature=True) + records.append({'file':path.name,'automatic_winner':index==1,'evaluation':dataclasses.asdict(state.evaluation), + 'home':[(str(n.start),str(n.end),n.pitch.number()) for n in state.home], + 'events':[(str(n.start),str(n.end),n.pitch.number()) for n in state.notes], + 'actions':[{'kind':a.kind,'start':str(a.start),'end':str(a.end),'contracts':[c.rule for c in a.contracts]} for a in state.actions]}) + value={'source_only':True,'automatic_winner':'candidate-1.xml','start':str(builder.start),'end':str(builder.end), + 'candidates':records,'frontiers':builder.frontiers,'timings':dict(builder.timings), + 'note':'Diagnostic candidates, not publication or evidence of exact reproduction.'} + (output/'candidates.json').write_text(json.dumps(value,indent=2,default=str)+'\n') + return value + + +def main(): + p=argparse.ArgumentParser(description=__doc__) + p.add_argument('--source',type=Path,required=True);p.add_argument('--role',choices=('acoustic_solo','electric_solo'),required=True) + p.add_argument('--data-dir',type=Path,default=Path('data'));p.add_argument('--output',type=Path,required=True) + p.add_argument('--cache-dir',type=Path);p.add_argument('--bars',type=int,default=4) + p.add_argument('--profile',action='store_true') + args=p.parse_args() + if args.bars<1:p.error('--bars must be positive') + with vocabulary.using(vocabulary.load(args.data_dir/'composition'),args.role): + b,document=build(args.source,args.role,args.data_dir,args.cache_dir) + ends=[x-b.analysis.body_offset for x in b.analysis.boundaries if x-b.analysis.body_offset>0] + b.end=min(b.end,ends[min(args.bars-1,len(ends)-1)]);b.operators.end=b.end + started=time.monotonic();seen=set() + def progress(stage,at,state,proposed,retained): + if stage=='frontier' and at.denominator==1 and at not in seen: + seen.add(at);print(f'frontier {at}: {len(proposed)} histories, {len(retained)} retained; {time.monotonic()-started:.1f}s',flush=True) + b.observe=progress + args.output.mkdir(parents=True,exist_ok=True) + profiler=cProfile.Profile() if args.profile else None + if profiler:profiler.enable() + try:b.compose() + finally: + if profiler:profiler.disable();profiler.dump_stats(args.output/'search.prof') + export(b,document,args.output) + print(json.dumps({'completed':len(b.completed),'seconds':b.timings['action_search_seconds'],'winner':str(b.winner_evaluation.total)})) + +if __name__=='__main__':main() diff --git a/audition/requirements-training.txt b/audition/requirements-training.txt new file mode 100644 index 0000000..6c43b66 --- /dev/null +++ b/audition/requirements-training.txt @@ -0,0 +1,2 @@ +# Development only. Runtime composition must not import this solver. +z3-solver==4.13.3.0 diff --git a/audition/structured_learning.py b/audition/structured_learning.py new file mode 100644 index 0000000..b856c37 --- /dev/null +++ b/audition/structured_learning.py @@ -0,0 +1,197 @@ +"""Development-only shared integer ranking from COMPLETE ordinary rollouts. + +The returned coefficients depend on the training examples. Target event arrays, +source identifiers and diagnostic oracle paths are never production artifacts. +""" +from collections import Counter +from fractions import Fraction as F +import itertools +import math +from generator import vocabulary +from generator.solo import selection + + +def contextual_candidates(selector,notes): + """Induce only whitelisted two/three-condition event contexts. + + These counts are training-dependent. No title, time index, song key, event + array, fingerprint or arbitrary predicate can appear in the rule schema. + """ + values=Counter() + for context,ticks in selector.contexts(notes): + terms=sorted((k,v) for k,v in context.items() if v in selection.CONDITIONS[k]) + for size in (2,3): + for terms_ in itertools.combinations(terms,size): + values['joint/'+'&'.join(k+'='+v for k,v in terms_)]+=ticks + return values + + +def vector(builder,actions,home=()): + e=builder.selector.complete(actions,home,start=builder.start,end=builder.end) + # Equal spans: clear normalization without rounding any musical time. + span=builder.end-builder.start + features={k:F(v)*span for k,v in e.raw.items() if not k.startswith('prior/') and v} + prior=sum(F(v)*span for k,v in e.contributions.items() if k.startswith('prior/')) + return features,prior + + +def constraint(role,gold,other,case): + left,lp=gold;right,rp=other + delta={k:left.get(k,0)-right.get(k,0) for k in left.keys()|right.keys()} + delta={k:v for k,v in delta.items() if v} + selection.validate_weights(dict.fromkeys(delta,0)) + return {'role':role,'delta':delta,'prior':lp-rp,'case':case} + + +def solve(rows,bound=1024): + """One shared coefficient table per role; deterministic pinned integer solver.""" + import z3 + if z3.get_version_string()!='4.13.3':raise ValueError('Use pinned z3-solver==4.13.3.0') + z3.set_param('smt.random_seed',0) + # Exact zero/one-coefficient cases need no combinatorial cardinality search. + # Checking every feature's integer interval proves the minimum support and + # L1 objective when all independent roles have such a solution. This does + # not claim the resulting sparse model reproduces any complete melody. + simple={} + for role in sorted({r['role'] for r in rows}): + subset=[r for r in rows if r['role']==role] + if all(r['prior']<=-1 for r in subset):simple[role]={};continue + choices=[] + for key in sorted({k for r in subset for k in r['delta']}): + low,high=-bound,bound + for r in subset: + slope=F(r['delta'].get(key,0));limit=-1-F(r['prior']) + if slope>0:high=min(high,math.floor(limit/slope)) + elif slope<0:low=max(low,math.ceil(limit/slope)) + elif limit<0:low,high=1,0;break + if low<=high: + value=low if low>0 else high if high<0 else 0 + choices.append((abs(value),key,value)) + if not choices:break + _,key,value=min(choices);simple[role]={key:value} + if set(simple)=={r['role'] for r in rows}: + for weights in simple.values():selection.validate_weights(weights) + return {'status':'solved','weights':simple,'bound':bound, + 'nonzero':sum(len(w) for w in simple.values()),'l1':sum(abs(v) for w in simple.values() for v in w.values()), + 'method':'exhaustive zero/one-coefficient integer intervals; minimum support and L1 within that support proven'} + names=sorted({(r['role'],k) for r in rows for k in r['delta']}) + variables={n:z3.Int('w_'+str(i)) for i,n in enumerate(names)} + limits=[v>=-bound for v in variables.values()]+[v<=bound for v in variables.values()] + rational=lambda v:z3.RealVal(str(F(v))) + constraints=[z3.Sum([variables[(r['role'],k)]*rational(v) for k,v in r['delta'].items()])+rational(r['prior'])<=-1 for r in rows] + check=z3.Solver();check.set(rlimit=2000000);check.add(*limits) + for i,c in enumerate(constraints):check.assert_and_track(c,'comparison_'+str(i)) + status=check.check() + if status==z3.unknown:return {'status':'solver_unknown','stage':'feasibility','reason':check.reason_unknown()} + if status==z3.unsat: + return {'status':'incompatible_with_current_features_and_bounds','bound':bound, + 'conflicts':[str(x) for x in check.unsat_core()], + 'identical_feature_pairs':[i for i,r in enumerate(rows) if not r['delta'] and r['prior']>=0]} + opt=z3.Optimize();opt.add(*limits,*constraints) + # Deterministic solver work budget. Exhaustion yields no proposed model; + # elapsed wall time never chooses a musical winner. + opt.set(rlimit=2000000) + opt.minimize(z3.Sum([z3.If(v==0,0,1) for v in variables.values()])) + opt.minimize(z3.Sum([z3.If(v<0,-v,v) for v in variables.values()])) + if opt.check()!=z3.sat:return {'status':'solver_unknown','reason':opt.reason_unknown()} + model=opt.model();weights={r:{} for r,_ in names} + for (role,key),v in variables.items(): + value=model.eval(v,model_completion=True).as_long() + if value:weights[role][key]=value + for w in weights.values():selection.validate_weights(w) + return {'status':'solved','weights':weights,'bound':bound, + 'nonzero':sum(len(w) for w in weights.values()),'l1':sum(abs(v) for w in weights.values() for v in w.values())} + + +def sounded(actions): + from generator.solo.types import sounding_notes + return tuple((n.start,n.end,n.pitch.number()) for n in sounding_notes(tuple(n for a in actions for n in a.notes))) + + +class CoverageTrace: + """Read-only observer. Targets diagnose cuts, never supply search decisions.""" + def __init__(self,gold): + self.gold=sounded(gold);self.records=[] + def matches(self,notes,end): + # These are committed whole actions, not a truncated playback preview. + # A quarter-note attack cannot reproduce a half-note target merely because + # its first quarter sounds the same. Future actions do not reopen it. + from generator.solo.types import sounding_notes + actual=tuple((n.start,n.end,n.pitch.number()) for n in sounding_notes(notes)) + expected=tuple((a,b,p) for a,b,p in self.gold if a