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

Construction

Browse Source
This commit is contained in:
khodowany
2026-10-07 23:45:17 -05:00
parent 3a01b4b6bd
commit 841a3f974e
8 changed files with 236 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
generator/comping_rhythm.py
+111 -17
View File
@@ -137,8 +137,8 @@ def validate_events(events, part, lengths=None):
raise ValueError('electric comping requires chord events')
for i, a in enumerate(ordered):
for b in ordered[i + 1:]:
if a.end > b.start and (part != 'acoustic' or a.lane == b.lane or 'combined' in (a.lane, b.lane)):
raise ValueError('comping attacks cannot overlap on the same strings')
if a.end > b.start:
raise ValueError('comping sound must release before the next attack')
def assign_acoustic(events, harmony, breaks=()):
"""Assign attacks, not notation fragments, across a complete source span.
@@ -164,10 +164,10 @@ def assign_acoustic(events, harmony, breaks=()):
else:
i += 1
result = [dataclasses.replace(e, lane=lane) for e, lane in zip(ordered, lanes)]
# Different strings may ring together, but never overlap a fresh attack on
# the same strings. Preserve the authored release whenever it is playable.
# One rhythm-guitar stream: every new attack releases the previous sound,
# including a change of lane. A combined attack has one shared release.
for i, a in enumerate(result):
following = next((b.start for b in result[i + 1:] if a.lane == b.lane or 'combined' in (a.lane, b.lane)), a.end)
following = result[i + 1].start if i + 1 < len(result) else a.end
result[i] = dataclasses.replace(a, end=min(a.end, following))
# Shortening must not strand an anticipation before its promised arrival.
# Rebind that event to the current chord, then reconsider its pair once.
@@ -325,7 +325,53 @@ def identity(pattern):
"""Layer swaps and anticipation permissions alone are not different rhythms."""
return tuple((fractions.Fraction(r['onset']), fractions.Fraction(r['duration'])) for r in pattern['events'])
def choose_cycles(groups, part, melody, seed):
def acoustic_context(groups, melody, breaks=()):
"""Immutable source timing for assigning and evaluating a complete stream.
Breaks are (source measure index, offset) pairs. Ordinary storage boundaries
are absent: adjacent figures must see each other's attacks before scoring.
"""
blocks, starts, cursor = [], {}, fractions.Fraction()
for group in groups:
start, context, lengths = cursor, [], []
for m, _, i in group:
starts[i] = cursor
context.extend((a + cursor - start, b + cursor - start) for a, b in melody[i])
lengths.append(generator.model.fraction(m['duration']))
cursor += lengths[-1]
starts[i + 1] = cursor
blocks.append((start, cursor, tuple(context), spans(flatten(group), None), tuple(lengths)))
full = tuple(row for group in groups for row in group)
return {'blocks': tuple(blocks), 'harmony': spans(flatten(full), None),
'breaks': tuple(sorted(starts[i] + t for i, t in breaks if i in starts))}
def realize_acoustic_form(rhythms, context):
"""The shared release/lane contract for selection, fills and final output.
Inputs retain authored releases. Re-evaluate those releases whenever an
edited neighboring figure changes the next attack; never restore duration
from previously shortened notation.
"""
if len(rhythms) != len(context['blocks']):
raise ValueError('acoustic realization requires one rhythm per source block')
events = []
for rhythm, (start, _, _, _, _) in zip(rhythms, context['blocks']):
events.extend(dataclasses.replace(e, start=e.start + start, end=e.end + start, harmonic_time=e.harmonic_time + start if e.harmonic_time is not None else None) for e in rhythm.events)
assigned = assign_acoustic(events, context['harmony'], context['breaks'])
result = []
for rhythm, (start, end, melody, harmony, lengths) in zip(rhythms, context['blocks']):
local = tuple(dataclasses.replace(e, start=e.start - start, end=e.end - start, harmonic_time=e.harmonic_time - start if e.harmonic_time is not None else None)
for e in assigned if start <= e.start < end)
exceptions = tuple(e for e in rhythm.exceptions if e[0] != 'layer_assignment')
result.append(dataclasses.replace(rhythm, events=local, contributions=evaluate(local, 'acoustic', melody, harmony, lengths),
pattern_id=rhythm.pattern_id.removesuffix('.anticipated') + ('.anticipated' if any(e.harmonic_time is not None for e in local) else ''),
exceptions=(*exceptions, ('layer_assignment', 'combined attacks or bass/upper eighth pairs; release at every next attack'))))
return tuple(result)
def form_cost(rhythms, groups):
return sum(r.cost * len(g) for r, g in zip(rhythms, groups))
def choose_cycles(groups, part, melody, seed, *, breaks=()):
"""Select a small recurring vocabulary against every occurrence in the form.
No local shortlist precedes whole-form evaluation. A pattern/mode is retained
@@ -349,12 +395,19 @@ def choose_cycles(groups, part, melody, seed):
requested = settings['cycle_patterns']
identities = {p['id']: identity(p) for p in patterns}
available = len(set(identities.values()))
cache = {}
cache, estimates = {}, {}
context = acoustic_context(run, melody, breaks) if part == 'acoustic' else None
for i, group in enumerate(run):
lengths = tuple(generator.model.fraction(m['duration']) for m, _, _ in group)
local_context = acoustic_context((group,), melody, breaks) if context else None
for p in patterns:
for anticipate in (False, True):
cache[i, p['id'], anticipate] = realize(project(p, fractions.Fraction(), lengths), group, part, melody, anticipate=anticipate)
value = realize(project(p, fractions.Fraction(), lengths), group, part, melody, anticipate=anticipate, assign=part != 'acoustic')
cache[i, p['id'], anticipate] = value
try:
estimates[i, p['id'], anticipate] = realize_acoustic_form((value,), local_context)[0] if context and value is not None else value
except ValueError:
estimates[i, p['id'], anticipate] = None
winner, examined = None, 0
for width in range(min(requested, len(run), available), 0, -1):
layers = []
@@ -362,7 +415,7 @@ def choose_cycles(groups, part, melody, seed):
unique = {}
for p in patterns:
for anticipate in (False, True):
values = [cache[i, p['id'], anticipate] for i in range(slot, len(run), width)]
values = [estimates[i, p['id'], anticipate] for i in range(slot, len(run), width)]
if any(v is None for v in values):
continue
cost = sum(v.cost * len(run[i]) for i, v in zip(range(slot, len(run), width), values))
@@ -383,7 +436,14 @@ def choose_cycles(groups, part, melody, seed):
if len({identities[o[1]['id']] for o in cycle}) != width:
continue
examined += 1
rank = (sum(o[0][0] for o in cycle), tuple(o[0][1] for o in cycle))
cost = sum(o[0][0] for o in cycle)
if context:
values = [cache[i, cycle[i % width][1]['id'], cycle[i % width][2]] for i in range(len(run))]
try:
cost = form_cost(realize_acoustic_form(values, context), run)
except ValueError:
continue
rank = (cost, tuple(o[0][1] for o in cycle))
if winner is None or rank < winner[0]:
winner = rank, cycle
if winner is not None:
@@ -395,16 +455,24 @@ def choose_cycles(groups, part, melody, seed):
'grouping': list(grouping), 'grouping_evidence': grouping_evidence, 'eligible_patterns': len(patterns),
'requested_patterns': requested, 'cycle_patterns': len(cycle), 'patterns': [o[1]['id'] for o in cycle],
'anticipation': [o[2] for o in cycle], 'cost': winner[0][0], 'complete_cycles_compared': examined,
'retrieval_cost': sum(o[0][0] for o in cycle),
'reduction': 'available distinct, legal patterns or source length' if len(cycle) < requested else None}
reports.append(report)
for i, group in enumerate(run):
_, p, anticipate = cycle[i % len(cycle)]
selected.append((p, anticipate, cache[i, p['id'], anticipate]))
locations.append({'cycle_start': run[0][0][2], 'cycle_slot': i % len(cycle), 'base_pattern': p['id'], 'fill': None})
add_fills(groups, selected, locations, part, melody, seed)
context = acoustic_context(groups, melody, breaks) if part == 'acoustic' else None
add_fills(groups, selected, locations, part, melody, seed, context=context)
if context:
realized = realize_acoustic_form([r for _, _, r in selected], context)
selected = [(p, a, r) for (p, a, _), r in zip(selected, realized)]
for report in reports:
report['final_cost'] = sum(r.cost * len(g) for g, r in zip(groups, realized) if report['source_start'] <= g[0][2] <= report['source_end'])
report['attack_assignment'] = 'sequential_combined_except_eighth_pairs'
return selected, locations, reports
def add_fills(groups, selected, locations, part, melody, seed):
def add_fills(groups, selected, locations, part, melody, seed, *, context=None):
"""Fixed source-position answers; no iteration seed or generated lead input."""
policy = generator.vocabulary.accompaniment(part)
spacing = policy['groove']['fill_spacing_measures']
@@ -415,7 +483,29 @@ def add_fills(groups, selected, locations, part, melody, seed):
options = []
if body_length < spacing:
return
for i, (group, (base, anticipate, current)) in enumerate(zip(groups, selected)):
original = realize_acoustic_form([r for _, _, r in selected], context) if context else tuple(r for _, _, r in selected)
def proposal(i, candidate, baseline):
if not context:
return candidate.cost < baseline[i].cost - 1e-9, baseline[i].cost - candidate.cost
raw = [r for _, _, r in selected]
raw[i] = candidate
try:
revised = realize_acoustic_form(raw, context)
except ValueError:
return False, 0
middle = generator.model.fraction(groups[i][0][0]['duration'])
# A second-bar fill must not alter the preceding/following block's lane
# assignment or release, nor any part of its own first bar.
if any(a.events != b.events for j, (a, b) in enumerate(zip(baseline, revised)) if j != i):
return False, 0
if tuple(e for e in baseline[i].events if e.start < middle) != tuple(e for e in revised[i].events if e.start < middle):
return False, 0
gain = form_cost(baseline, groups) - form_cost(revised, groups)
return gain > 1e-9, gain
for i, (group, (base, anticipate, _)) in enumerate(zip(groups, selected)):
current = original[i]
if len(group) != 2:
continue
last, _, index = group[-1]
@@ -434,17 +524,21 @@ def add_fills(groups, selected, locations, part, melody, seed):
hybrid = {**base, 'id': base['id'] + '.fill.' + variation['id'], 'events':
[r for r in base['events'] if fractions.Fraction(r['onset']) < middle] +
[r for r in variation['events'] if fractions.Fraction(r['onset']) >= middle]}
candidate = realize(hybrid, group, part, melody, anticipate=anticipate)
if candidate is None or candidate.cost >= current.cost - 1e-9:
candidate = realize(hybrid, group, part, melody, anticipate=anticipate, assign=part != 'acoustic')
if candidate is None:
continue
if tuple(e for e in candidate.events if e.start < middle) != tuple(e for e in current.events if e.start < middle):
if not context and tuple(e for e in candidate.events if e.start < middle) != tuple(e for e in current.events if e.start < middle):
continue
valid, gain = proposal(i, candidate, original)
if not valid:
continue
gain = current.cost - candidate.cost
options.append((-gain, generator.solo.policy.stable_rank(seed, (index, hybrid['id'])), i, index, hybrid, anticipate, candidate, reason))
chosen = []
for _, _, i, index, pattern, anticipate, candidate, reason in sorted(options):
if any(min(abs(index - other), body_length - abs(index - other)) < spacing for other in chosen):
continue
if context and not proposal(i, candidate, realize_acoustic_form([r for _, _, r in selected], context))[0]:
continue
selected[i] = pattern, anticipate, candidate
locations[i]['fill'] = {'pattern': pattern['id'], 'source_index': index, 'reason': reason}
chosen.append(index)