khodo/product-vamp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Comp

Browse Source
This commit is contained in:
khodowany
2026-10-01 16:41:18 -05:00
parent 28fc67fba1
commit f5861fcdf6
22 changed files with 1473 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
generator/solo/realization.py
+190 -72
View File
@@ -6,6 +6,8 @@ import functools
import time
import generator.solo.drafts
import generator.solo.references
import generator.solo.routes
import generator.fretboard
import generator.model
@@ -354,6 +356,8 @@ class SearchContext:
resources: dict = dataclasses.field(default_factory=dict)
revise: bool = True
batch: int = 0
references: tuple = ()
stage: str = 'elaborated'
diagnostics: dict = dataclasses.field(default_factory=lambda: {'searches': 0, 'cache_hits': 0, 'generated_candidates': 0, 'evaluated_extensions': 0, 'peak_states': 0, 'candidate_families': {}, 'surviving_families': {}})
def note_family(notes):
@@ -421,19 +425,9 @@ def draft_pitch_paths(rhythm, plan, choices, harmony, sequences, incoming, conte
expected = prior.pitch.number() + step if prior else rhythm.reference.notes[0].pitch.number()
anchor = next_anchor[index]
target = anchor.pitch.number() if anchor else expected
route = expected
if prior and anchor and span and span.scale:
try:
start_degree = generator.solo.types.scale_index(span.scale, prior.pitch.number())
target_degree = generator.solo.types.scale_index(span.scale, target)
displacement = target_degree - start_degree
count = min(16, max(2, remaining + 1))
paths = sequences.retrieve(count, displacement, (displacement > 0) - (displacement < 0))
if paths:
route = generator.solo.types.scale_pitch(span.scale, start_degree + paths[0].steps[0]).number()
except ValueError:
# Collection changes preserve the incoming sound, never its old index.
route = prior.pitch.number() + ((target > prior.pitch.number()) - (target < prior.pitch.number())) * 2
# Theme invention follows a complete paired rhythmic/interval idea.
# Authored sequence routes are instantiated as whole alternatives later.
route = expected if not anchor else target
ranked = sorted(available, key=lambda p: (abs(p.number() - expected), p))[:3]
if available:
bridge = min(available, key=lambda p: (abs(p.number() - route), p))
@@ -443,14 +437,13 @@ def draft_pitch_paths(rhythm, plan, choices, harmony, sequences, incoming, conte
direction = (pitch.number() > prior.pitch.number()) - (pitch.number() < prior.pitch.number()) if prior else 0
intended = (step > 0) - (step < 0)
identifying = bool(local_index in (1, 2) and group.operation not in ('close', 'answer') and not slot.forced)
if identifying and reference and direction != intended:
continue
jump = abs(pitch.number() - prior.pitch.number()) if prior else 0
stopping = slot.end - slot.start >= generator.solo.policy.DEFAULT.stopping_note
motion_cost = abs(pitch.number() - expected) / 12 * (1.5 if identifying else 0.35)
direction_cost = max(0, abs(pitch.number() - target) - 2 * max(0, remaining - 1)) / 12 if anchor else 0
exposure = 0.6 if stopping and span and pitch.number() % 12 not in span.roles.basic and not quoted else 0
score = cost + motion_cost + direction_cost + max(0, jump - 5) / 12 + exposure
supported_leap = reference and abs(step) >= 3 and abs(pitch.number() - expected) <= 2
score = cost + motion_cost + direction_cost + (0 if supported_leap else max(0, jump - 5) / 12) + exposure
sources = reference[min(slot.reference, len(reference) - 1)].sources if reference else ()
note = generator.solo.types.Note(slot.start, slot.end, pitch, sources, 'quote' if quoted else 'motif', event_id=slot.id)
path = (*notes, note)
@@ -489,10 +482,9 @@ def make_draft(rhythm, notes, plan, harmony, incoming):
identity = make_identity(group.id, parent, actual, start, end, group.operation, harmony)
if home is None:
home = identity
structural.extend(n.event_id for n in actual[:3])
directions = tuple((b.pitch.number() > a.pitch.number()) - (b.pitch.number() < a.pitch.number()) for a, b in zip(actual, actual[1:]))
# Closing simplification explicitly releases the characteristic core.
preserves = () if group.operation == 'close' else ('rhythm', 'opening_contour')
preserves = ()
if plan.treatment == 'rhythmic_reference':
preserves = ('rhythm',)
transforms.append(generator.solo.types.Transformation(group.id, parent, home.id, group.operation, start, end, tuple(n.event_id for n in actual), tuple((n.start - start, n.end - start) for n in actual), directions, preserves, group.source))
@@ -503,7 +495,7 @@ def make_draft(rhythm, notes, plan, harmony, incoming):
if note:
structural.append(note.event_id)
joins = tuple(t.start for t in transforms[1:])
return generator.solo.types.MelodicDraft(rhythm.id, plan.start, plan.end, notes, tuple(transforms), tuple(dict.fromkeys(structural)), tuple(protected), rhythm.anchors, home, rhythm.relationship, joins)
return generator.solo.types.MelodicDraft(rhythm.id, plan.start, plan.end, notes, tuple(transforms), tuple(dict.fromkeys(structural)), tuple(protected), rhythm.anchors, home, rhythm.relationship, joins, reference=rhythm.reference.id, reference_motif=rhythm.reference)
def checked_phrase(plan, draft, notes, contracts, incoming, harmony, tempo, edits=()):
valid, _ = generator.solo.drafts.contract_valid(draft, notes)
@@ -524,14 +516,17 @@ def checked_phrase(plan, draft, notes, contracts, incoming, harmony, tempo, edit
return None
shared = bool(plan.passage_id and incoming.passage_id == plan.passage_id)
score_notes = canonical if shared or notes and notes[0].continuation else generator.solo.drafts.local_sounding(notes)
evaluation = generator.solo.scoring.Evaluation(evaluation_plan(plan, draft.anchors), harmony.choices, plan.end)
evaluation_key = (plan, draft.anchors)
if evaluation_key not in harmony.evaluations:
harmony.evaluations[evaluation_key] = generator.solo.scoring.Evaluation(evaluation_plan(plan, draft.anchors), harmony.choices, plan.end)
evaluation = harmony.evaluations[evaluation_key]
if not evaluation.feasible(score_notes):
return None
features = evaluation.features(score_notes, previous=incoming.last if not shared else None, movement=travel + (incoming.travel if shared else 0))
features.update(generator.solo.drafts.relationship_features(draft, notes))
# Fidelity is checked against explicit transformed identities, not against
# the source head as one undifferentiated phrase-long quotation.
features['fidelity'] = features['development']
features.pop('fidelity', None)
features['harmonic_exposure'] = checked.exposure
intended_repeats = 0
if draft.home:
@@ -589,32 +584,17 @@ def emitted_notes(phrase):
def window_variants(phrase, low, high, harmony, sequences, kind):
"""Two fixed outside endpoints bound edits to the existing melody."""
notes = emitted_notes(phrase)
protected = set((*phrase.draft.structural, *phrase.draft.protected))
protected = set((*phrase.draft.structural, *phrase.draft.protected, *(e for f in phrase.draft.features for e in f.events), *(n.event_id for r in phrase.routes for n in r.notes)))
inside = [i for i, n in enumerate(notes) if low <= n.start and n.end <= high]
editable = [i for i in inside if notes[i].event_id not in protected]
if not editable:
return ()
candidates = []
left, right = max(0, inside[0] - 1), min(len(notes) - 1, inside[-1] + 1)
# Route both sides of a join as one path, with the original outside pitches.
if 1 < right - left < 16:
scale = harmony.at(notes[left].start).scale
try:
first = generator.solo.types.scale_index(scale, notes[left].pitch.number())
displacement = generator.solo.types.scale_index(scale, notes[right].pitch.number()) - first
paths = sequences.retrieve(right - left + 1, displacement, (displacement > 0) - (displacement < 0))
except ValueError:
paths = ()
for path in paths[:generator.solo.policy.DEFAULT.join_alternatives]:
updated = list(notes)
for i in editable:
offset = path.offsets()[i - left]
intended = generator.solo.types.scale_pitch(scale, first + offset).number()
available = options(harmony.choices, notes[i].start, notes[i].end, phrase.plan.register)
if available:
pitch = min(available, key=lambda p: (abs(p.number() - intended), p))
updated[i] = dataclasses.replace(notes[i], pitch=pitch, position=None)
candidates.append((tuple(updated), 'sequence ' + ' + '.join(path.sources)))
# Compare complete routes through both fixed endpoints. Reject an invalid
# path as a unit; never substitute nearest pitches for individual steps.
for route in generator.solo.routes.candidates(notes[left:right + 1], protected, harmony, sequences, phrase.plan.register):
candidates.append(((*notes[:left], *route.notes, *notes[right + 1:]), 'complete route ' + ' + '.join(route.sources)))
for index in editable:
note = notes[index]
if index + 1 < len(notes) and notes[index + 1].start <= high:
@@ -625,20 +605,6 @@ def window_variants(phrase, low, high, harmony, sequences, kind):
# The tail and pickup move together, preserving total sounding time.
time = after.start - fractions.Fraction(1, 2)
candidates.append(((*notes[:index], dataclasses.replace(note, end=time), dataclasses.replace(after, start=time), *notes[index + 2:]), 'reopen tail into pickup'))
if kind == 'elaboration' and index + 1 < len(notes) and fractions.Fraction(1) <= note.end - note.start < 2:
# Add motion only as an alternative evaluated against the unchanged
# activity envelope; keeping the original draft is always available.
split = note.end - fractions.Fraction(1, 2)
scale = harmony.at(split).scale
target = notes[index + 1].pitch
try:
degree = generator.solo.types.scale_index(scale, note.pitch.number())
finish = generator.solo.types.scale_index(scale, target.number())
pitch = generator.solo.types.scale_pitch(scale, degree + ((finish > degree) - (finish < degree)))
addition = generator.solo.types.Note(split, note.end, pitch, note.sources, 'elaboration', event_id=note.event_id + '_passing')
candidates.append(((*notes[:index], dataclasses.replace(note, end=split), addition, *notes[index + 1:]), 'interpolate toward next structural event'))
except ValueError:
pass
return tuple(candidates)
def improve_window(phrase, incoming, harmony, sequences, tempo, low, high, kind, context):
@@ -648,7 +614,7 @@ def improve_window(phrase, incoming, harmony, sequences, tempo, low, high, kind,
# Keep distinct rhythmic and pitch revisions; no generic unlimited search.
proposals = generator.solo.sequences.diverse(proposals, lambda p: (abs(len(p[0]) - len(original)), p[1], tuple(n.pitch for n in p[0])), lambda p: (p[1].split()[0], len(p[0])), generator.solo.policy.DEFAULT.join_alternatives)
for notes, source in proposals:
if notes == original:
if tuple((n.start, n.end, n.pitch, n.continuation) for n in notes) == tuple((n.start, n.end, n.pitch, n.continuation) for n in original):
continue
changed_before = tuple(n.event_id for n in original if n not in notes)
changed_after = tuple(n.event_id for n in notes if n not in original)
@@ -656,16 +622,155 @@ def improve_window(phrase, incoming, harmony, sequences, tempo, low, high, kind,
revised = checked_phrase(phrase.plan, phrase.draft, notes, phrase.state.treatments, incoming, harmony, tempo, (*phrase.edits, edit))
context.diagnostics[kind + '_attempts'] = context.diagnostics.get(kind + '_attempts', 0) + 1
if revised:
viable.append(revised)
viable.append(dataclasses.replace(revised, initial=phrase.initial, refined=phrase.refined, routes=phrase.routes))
chosen = min(viable, key=lambda p: (p.score, len(p.edits), tuple((n.pitch, n.start, n.end) for n in emitted_notes(p))))
if chosen is not phrase:
context.diagnostics[kind + '_accepted'] = context.diagnostics.get(kind + '_accepted', 0) + 1
return chosen
def rebase_draft(draft, notes, harmony):
"""Reassign edited events by absolute group time, retaining source contracts."""
notes = tuple(dataclasses.replace(n, position=None) for n in notes)
groups = []
original_groups = {g.id: g for g in draft.transformations}
for i, group in enumerate(draft.transformations):
low = draft.start if i == 0 else group.start
high = draft.transformations[i + 1].start if i + 1 < len(draft.transformations) else draft.end
actual = tuple(n for n in notes if low <= n.start < high)
if actual:
parent = group.parent
while parent in original_groups and parent not in {g.id for g in groups}:
parent = original_groups[parent].parent
groups.append(dataclasses.replace(group, parent=parent, events=tuple(n.event_id for n in actual), rhythm=group.rhythm if group.preserves else tuple((n.start - group.start, n.end - group.start) for n in actual)))
home = draft.home
if home and groups and home.id == groups[0].id:
actual = tuple(n for n in notes if n.event_id in groups[0].events)
home = make_identity(home.id, home.parent, actual, groups[0].start, groups[0].end, home.relationship, harmony)
return dataclasses.replace(draft, notes=notes, transformations=tuple(groups), home=home)
def replace_idea(phrase, donor, low, high, incoming, harmony, tempo):
"""A whole two-bar idea is editable until its identifying features are selected."""
original = emitted_notes(phrase)
if any(n.start < low < n.end or n.start < high < n.end for n in original):
return None
replacement = tuple(n for n in emitted_notes(donor) if low <= n.start and n.end <= high)
if not replacement or tuple((n.start, n.end, n.pitch) for n in replacement) == tuple((n.start, n.end, n.pitch) for n in original if low <= n.start < high):
return None
protected = set((*phrase.draft.structural, *phrase.draft.protected))
identifiers = {}
for before in original:
if before.event_id in protected and low <= before.start < high:
after = next((n for n in replacement if (n.start, n.end, n.pitch) == (before.start, before.end, before.pitch)), None)
if not after:
return None
identifiers[after.event_id] = before.event_id
replacement = tuple(dataclasses.replace(n, event_id=identifiers.get(n.event_id, n.event_id + '_replacement'), position=None) for n in replacement)
notes = tuple(n for n in original if n.end <= low) + replacement + tuple(n for n in original if n.start >= high)
valid, _ = generator.solo.drafts.contract_valid(phrase.draft, notes)
if not valid:
return None
draft = rebase_draft(phrase.draft, notes, harmony)
edit = generator.solo.types.DraftEdit('replace_idea', low, high, tuple(n.event_id for n in original if low <= n.start < high), tuple(n.event_id for n in replacement), donor.draft.reference + '; ' + donor.draft.id)
return checked_phrase(phrase.plan, draft, notes, phrase.state.treatments, incoming, harmony, tempo, (*phrase.edits, edit))
def refine(phrase, pool, incoming, harmony, sequences, tempo, context, search=True):
initial = phrase.draft
for low in (phrase.plan.start, phrase.plan.start + 8) if search else ():
high = min(phrase.plan.end, low + 8)
if high - low < 8 or phrase.draft.protected:
continue
viable = [phrase]
for donor in pool:
if donor.draft.id == phrase.draft.id:
continue
context.diagnostics['idea_replacements_tried'] = context.diagnostics.get('idea_replacements_tried', 0) + 1
revised = replace_idea(phrase, donor, low, high, incoming, harmony, tempo)
if revised:
viable.append(revised)
chosen = min(viable, key=lambda p: (p.score, len(p.edits), p.draft.id))
if chosen is not phrase:
context.diagnostics['idea_replacements_selected'] = context.diagnostics.get('idea_replacements_selected', 0) + 1
phrase = chosen
if search and context.revise:
phrase = revise_joins(phrase, incoming, harmony, sequences, tempo, context)
notes = emitted_notes(phrase)
draft = rebase_draft(phrase.draft, notes, harmony)
# Only now commit a selected, inspectable relationship for elaboration.
features = generator.solo.references.identifying_features(notes, draft.reference)
draft = dataclasses.replace(draft, features=features)
# Rebuild lineage after group removal or replacement, including when no run
# is selected afterward. The introduced home is the actual refined tune.
rebuilt = checked_phrase(phrase.plan, draft, notes, phrase.state.treatments, incoming, harmony, tempo, phrase.edits)
if rebuilt is None:
raise ValueError(phrase.plan.id + ': refined theme lost a committed contract')
return dataclasses.replace(rebuilt, initial=initial, refined=draft)
def elaborate(phrase, incoming, harmony, sequences, tempo, context):
started = time.monotonic()
for transform in phrase.draft.transformations:
phrase = improve_window(phrase, incoming, harmony, sequences, tempo, transform.start, transform.end, 'elaboration', context)
protected = set((*phrase.draft.structural, *phrase.draft.protected, *(e for f in phrase.draft.features for e in f.events)))
if phrase.draft.protected:
return phrase
# Elaborate selected approaches in the existing tune, at most two per phrase.
notes = emitted_notes(phrase)
windows = []
for right in range(2, len(notes)):
if notes[right].event_id not in protected and right != len(notes) - 1:
continue
left = next((i for i in range(right - 1, -1, -1) if notes[right].start - notes[i].start >= 4), None)
if left is not None and notes[right].start - notes[left].start <= 4:
windows.append((notes[left].start, notes[right].start))
for low, high in windows[-2:]:
if any(low < r.destination.end and high > r.notes[0].start for r in phrase.routes):
continue
notes = emitted_notes(phrase)
left = next(i for i, n in enumerate(notes) if n.start == low)
right = next(i for i, n in enumerate(notes) if n.start == high)
patterns = [notes[left:right + 1]]
# A full approach rhythm, preserving every committed event. Existing
# silence remains a candidate; short-note motion is never compulsory.
moving = []
timepoint = low
while timepoint < high:
original = next((n for n in notes[left:right] if n.start == timepoint), None)
end = min(high, timepoint + fractions.Fraction(1, 2))
moving.append(dataclasses.replace(original, end=end, position=None) if original else generator.solo.types.Note(timepoint, end, notes[left].pitch, role='run', event_id=f'route_{phrase.draft.id}_{timepoint}'))
timepoint = end
patterns.append((*moving, notes[right]))
for reference in context.references:
for a, b in reference.runs:
if high - low != b - a:
continue
profile = tuple(n for n in reference.decorated if a <= n.start and n.end <= b)
timed = []
for index, event in enumerate(profile):
start, end = low + event.start - a, low + event.end - a
original = next((n for n in notes[left:right] if n.start == start), None)
timed.append(dataclasses.replace(original, end=end, position=None) if original else generator.solo.types.Note(start, end, notes[left].pitch, (reference.id,), 'run', event_id=f'profile_{phrase.draft.id}_{start}'))
if timed:
patterns.append((*timed, notes[right]))
viable = [phrase]
for pattern in dict.fromkeys(patterns):
# Fixed events retain their duration and attack identity as well as pitch.
if any(not any((n.event_id, n.start, n.end) == (p.event_id, p.start, p.end) for p in pattern) for n in notes[left:right + 1] if n.event_id in protected):
continue
for route in generator.solo.routes.candidates(pattern, protected, harmony, sequences, phrase.plan.register, 'approach the theme destination'):
actual = (*notes[:left], *route.notes, *notes[right + 1:])
context.diagnostics['complete_routes_tried'] = context.diagnostics.get('complete_routes_tried', 0) + 1
if not generator.solo.drafts.contract_valid(phrase.draft, actual)[0]:
continue
draft = rebase_draft(phrase.draft, actual, harmony)
edit = generator.solo.types.DraftEdit('complete_route', low, notes[right].end, tuple(n.event_id for n in notes[left:right + 1]), tuple(n.event_id for n in route.notes), ' + '.join(route.sources))
revised = checked_phrase(phrase.plan, draft, actual, phrase.state.treatments, incoming, harmony, tempo, (*phrase.edits, edit))
if revised:
# Retain the refined theme, not the ornamental version, as home.
state = dataclasses.replace(revised.state, home=phrase.state.home, themes=phrase.state.themes)
viable.append(dataclasses.replace(revised, state=state, initial=phrase.initial, refined=phrase.refined, routes=(*phrase.routes, route)))
else:
context.diagnostics['complete_routes_rejected'] = context.diagnostics.get('complete_routes_rejected', 0) + 1
chosen = min(viable, key=lambda p: (p.score, len(p.edits), tuple(n.pitch for n in emitted_notes(p))))
if chosen is not phrase:
context.diagnostics['complete_routes_selected'] = context.diagnostics.get('complete_routes_selected', 0) + 1
phrase = chosen
context.diagnostics['elaboration_seconds'] = context.diagnostics.get('elaboration_seconds', 0) + time.monotonic() - started
return phrase
@@ -674,7 +779,15 @@ def revise_joins(phrase, incoming, harmony, sequences, tempo, context):
width = fractions.Fraction(str(generator.solo.policy.DEFAULT.join_window))
for _ in range(generator.solo.policy.DEFAULT.revision_sweeps):
before = phrase
for join in phrase.draft.joins:
notes = emitted_notes(phrase)
def weakness(join):
before = next((n for n in reversed(notes) if n.start < join), None)
after = next((n for n in notes if n.start >= join), None)
return (max(0, abs(after.pitch.number() - before.pitch.number()) - 4) + float(max(0, after.start - before.end))) if before and after else 0
joins = sorted(phrase.draft.joins, key=lambda j: (-weakness(j), j))[:generator.solo.policy.DEFAULT.revision_joins]
for join in joins:
phrase = improve_window(phrase, incoming, harmony, sequences, tempo, max(phrase.plan.start, join - width), min(phrase.plan.end, join + width), 'join', context)
if phrase == before:
break
@@ -683,7 +796,7 @@ def revise_joins(phrase, incoming, harmony, sequences, tempo, context):
def revise_pair(first, second, incoming, choices, sequences, tempo, context):
"""Revise an uncommitted connected phrase seam, then rebuild both states."""
if not context.revise or not first.draft or not second.draft or first.plan.ending != 'connected' or first.plan.passage_id != second.plan.passage_id:
if first.routes or second.routes or first.refined or second.refined or not context.revise or not first.draft or not second.draft or first.plan.ending != 'connected' or first.plan.passage_id != second.plan.passage_id:
return first, second
incoming = incoming if isinstance(incoming, generator.solo.types.ContinuationState) else generator.solo.types.ContinuationState('', last=incoming)
harmony = generator.solo.harmonic.Harmony(choices)
@@ -722,8 +835,8 @@ def alternatives(plan, measures, choices, sequences, rhythms, tempo, previous=No
# library cannot accidentally reuse the identity of a collected object.
context.resources[id(sequences)] = sequences
context.resources[id(rhythms)] = rhythms
pool_key = (plan, previous, generator.solo.policy.stable_rank('harmonic-cache', choices), tempo, id(sequences), id(rhythms))
key = (*pool_key, context.revise)
pool_key = (plan, previous, generator.solo.policy.stable_rank('harmonic-cache', choices), tempo, id(sequences), id(rhythms), tuple(r.version for r in context.references))
key = (*pool_key, context.revise, context.stage)
if key in context.results:
context.diagnostics['cache_hits'] += 1
return context.results[key]
@@ -736,10 +849,11 @@ def alternatives(plan, measures, choices, sequences, rhythms, tempo, previous=No
selected = context.pools[pool_key]
context.diagnostics['reused_draft_pools'] = context.diagnostics.get('reused_draft_pools', 0) + 1
else:
constructions = generator.solo.drafts.rhythms(plan, incoming, measures, choices, rhythms, context.batch)
constructions = generator.solo.drafts.rhythms(plan, incoming, measures, choices, rhythms, context.batch, context.references)
complete, failure = [], ''
for rhythm in constructions:
paths, error = draft_pitch_paths(rhythm, plan, choices, harmony, sequences, incoming, context)
candidate_plan = generator.solo.drafts.ending_plan(plan, rhythm.anchors)
paths, error = draft_pitch_paths(rhythm, candidate_plan, choices, harmony, sequences, incoming, context)
failure = error or failure
context.diagnostics['generated_candidates'] += len(paths)
for notes, cost, contracts in paths:
@@ -750,29 +864,33 @@ def alternatives(plan, measures, choices, sequences, rhythms, tempo, previous=No
if notes and incoming.last and incoming.last.end == notes[0].start and incoming.last.pitch == notes[0].pitch and notes[0].role != 'quote' and not any(a.onset_required and a.earliest <= notes[0].start <= a.latest for a in rhythm.anchors):
realizations.append((dataclasses.replace(notes[0], continuation=True), *notes[1:]))
for actual in realizations:
draft = make_draft(rhythm, actual, plan, harmony, incoming)
phrase = checked_phrase(plan, draft, actual, contracts, incoming, harmony, tempo)
draft = make_draft(rhythm, actual, candidate_plan, harmony, incoming)
phrase = checked_phrase(candidate_plan, draft, actual, contracts, incoming, harmony, tempo)
if phrase:
complete.append(dataclasses.replace(phrase, score=phrase.score + 0.1 * cost))
complete.append(phrase)
if not complete:
raise SearchExhausted(f'{plan.id}: bounded phrase draft search exhausted; {failure or "no complete draft satisfies timing, anchors, treatments and motif contracts"}; {len(constructions)} constructions / {BEAM_WIDTH} pitch states (not proof that the phrase is unplayable)')
context.diagnostics['construction_seconds'] = context.diagnostics.get('construction_seconds', 0) + time.monotonic() - started
selected = generator.solo.sequences.diverse(complete, lambda p: (p.score, p.draft.id, tuple(n.pitch for n in p.draft.notes)), lambda p: (p.draft.relationship, note_family(p.draft.notes), p.anchors[-1].duration if p.anchors else None), generator.solo.policy.DEFAULT.phrase_alternatives)
selected = tuple(elaborate(p, incoming, harmony, sequences, tempo, context) for p in selected)
context.pools[pool_key] = selected
input_id = generator.solo.policy.stable_rank(plan.seed, (plan, previous, choices))[:16]
pool_id = generator.solo.policy.stable_rank(plan.seed, tuple((p.draft.id, emitted_notes(p)) for p in selected))
context.diagnostics.setdefault('draft_pool_fingerprints', {})[input_id] = pool_id
result = []
for phrase in selected:
if context.revise:
phrase = revise_joins(phrase, incoming, harmony, sequences, tempo, context)
for index, phrase in enumerate(selected):
if context.stage == 'initial':
phrase = dataclasses.replace(phrase, initial=phrase.draft)
else:
phrase = refine(phrase, selected, incoming, harmony, sequences, tempo, context, search=index < generator.solo.policy.DEFAULT.refinement_alternatives)
if context.stage == 'elaborated':
phrase = elaborate(phrase, incoming, harmony, sequences, tempo, context)
result.append(phrase)
result = tuple(sorted(result, key=lambda p: (p.score, p.draft.id)))
for phrase in result:
family = note_family(emitted_notes(phrase))
counts = context.diagnostics['surviving_families']
counts[family] = counts.get(family, 0) + 1
context.diagnostics.setdefault('complete_ideas', {})[input_id] = [{'id': p.draft.id, 'reference': p.draft.reference, 'score': round(p.score, 6), 'attacks': len(emitted_notes(p)), 'ending': p.plan.ending, 'hold': str(p.anchors[-1].duration) if p.anchors else '', 'edits': [e.kind for e in p.edits]} for p in result]
context.results[key] = result
return result