product-vamp/tests/test_comping_cycles.py

"""Recurring source-position grooves, fills and chord-first bass behavior."""
import generator.composition_types
import generator.composition
import generator.source
import generator.validation
import concurrent.futures
import copy
import dataclasses
import fractions
import multiprocessing
import pathlib
import tempfile
import unittest
import unittest.mock

import generator.accompaniment
import generator.acoustic
import generator.arrange.accompaniment
import generator.arrange.form
import generator.comping
import generator.comping_rhythm
import generator.fretboard
import generator.harmony
import generator.model
import generator.production
import generator.score
import generator.vocabulary
import tests.solo_fixtures

def groups(source):
  measures = source['staves'][0]['measures']
  carried = generator.accompaniment.carried_chords(measures)
  return [tuple((m, carried[i], i) for i, m in enumerate(measures[a:a + 2], a)) for a in range(0, len(measures), 2)]

def signature(measure):
  return [(generator.model.fraction(n['onset']), generator.model.fraction(n['duration']),
    [(generator.model.pitch_number(p), p.get('tablature'), bool(p.get('tie_from')), bool(p.get('tie_to'))) for p in n['pitches']])
    for n in generator.score.measure_notes(measure)]

class RecurringGrooves(unittest.TestCase):
  def test_cycles_repeat_distinct_patterns_for_arbitrary_lengths(self):
    for count in (9, 18, 32):
      source = tests.solo_fixtures.song(count=count, chords=('C7',))
      for part, width in (('acoustic', 2), ('electric', 4)):
        canonical = generator.comping.Canonical(source, part, 'cycle')
        self.assertEqual(canonical.groove_report[0]['cycle_patterns'], width)
        patterns = [p for p, _ in canonical.patterns.values()]
        identities = [generator.comping_rhythm.identity(p) for p in patterns[:width]]
        self.assertEqual(len(set(identities)), width)
        for i, p in enumerate(patterns):
          self.assertEqual(generator.comping_rhythm.identity(p), identities[i % width])
        # The odd final measure is the first half of its cycle's figure.
        if count % 2:
          self.assertEqual(canonical.block_rhythms[count - 1].events,
            tuple(e for e in canonical.block_rhythms[(len(patterns) - 1) % width * 2].events if e.start < 4))
  def test_whole_form_evaluation_can_reject_locally_best_pattern(self):
    source = tests.solo_fixtures.song(count=8, chords=('C7',))
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['electric']
    policy['groove']['cycle_patterns'] = 1
    policy['patterns'] = [p for p in policy['patterns'] if p['id'] in ('electric.charleston.statement', 'electric.quarters.statement')]
    realize = generator.comping_rhythm.realize
    def evaluated(pattern, inputs, *args, **kwargs):
      candidate = realize(pattern, inputs, *args, **kwargs)
      # Charleston wins the first two bars, but loses the full eight-bar span.
      cost = (-1 if inputs[0][2] == 0 else 2) if 'charleston' in pattern['id'] else 0
      return dataclasses.replace(candidate, contributions=(('test_context', cost),))
    with generator.vocabulary.using(resources, 'electric_accompaniment'), unittest.mock.patch('generator.comping_rhythm.realize', side_effect=evaluated):
      selected, _, report = generator.comping_rhythm.choose_cycles(groups(source), 'electric', generator.comping_rhythm.melody_contexts(source), 'coverage')
    self.assertTrue(all(p['id'] == 'electric.quarters.statement' for p, _, _ in selected))
    self.assertEqual(report[0]['cost'], 0)
  def test_acoustic_cycle_is_ranked_after_cross_block_assignment(self):
    source = tests.solo_fixtures.song(count=4, chords=('C7',))
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['acoustic']
    policy['groove']['cycle_patterns'] = 1
    policy['patterns'] = [{'id': name, 'family': 'quarters', 'lengths': ['4', '4'], 'events': [
      {'onset': t, 'duration': '1', 'layer': 'acoustic'} for t in times]}
      for name, times in (('crossing', ('0', '1/2', '15/2')), ('local', ('0', '2', '6')))]
    def evaluated(events, *args):
      # Isolated retrieval prefers 'local' (0) over 'crossing' (1).
      # The complete stream permits a prepared upper entry in block two.
      cost = -3 if events[0].lane == 'upper' else 1 if events[1].start == fractions.Fraction(1, 2) else 0
      return [('test_entry', cost)]
    with generator.vocabulary.using(resources, 'acoustic_accompaniment'), unittest.mock.patch('generator.comping_rhythm.evaluate', side_effect=evaluated):
      selected, _, reports = generator.comping_rhythm.choose_cycles(groups(source), 'acoustic', generator.comping_rhythm.melody_contexts(source), 'complete')
    self.assertEqual([p['id'] for p, _, _ in selected], ['crossing', 'crossing'])
    self.assertEqual([r.events[0].lane for _, _, r in selected], ['bass', 'upper'])
    self.assertEqual(reports[0]['retrieval_cost'], 4)
    self.assertEqual(reports[0]['cost'], -4)
    self.assertEqual(reports[0]['cost'], sum(r.cost * 2 for _, _, r in selected))
    self.assertEqual(reports[0]['cost'], reports[0]['final_cost'])
    for _, _, rhythm in selected:
      generator.comping_rhythm.validate_events(rhythm.events, 'acoustic')
  def test_acoustic_fill_cannot_reassign_neighboring_attacks(self):
    source = tests.solo_fixtures.song(count=8, chords=('C7',))
    source['staves'][0]['measures'][2]['section_label'] = 'Source boundary'
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['acoustic']
    policy['patterns'] = [{'id': name, 'family': 'quarters', 'lengths': ['4', '4'], 'events': [
      {'onset': t, 'duration': '1/2', 'layer': 'acoustic'} for t in ('0', '2', last)]}
      for name, last in (('base', '5'), ('unsafe', '15/2'), ('safe', '7'))]
    source_groups = groups(source)
    melody = generator.comping_rhythm.melody_contexts(source)
    context = generator.comping_rhythm.acoustic_context(source_groups, melody)
    base = policy['patterns'][0]
    selected = [(base, False, generator.comping_rhythm.realize(base, g, 'acoustic', melody, assign=False)) for g in source_groups]
    locations = [{'fill': None} for _ in selected]
    original = generator.comping_rhythm.realize_acoustic_form([r for _, _, r in selected], context)
    def evaluated(events, part, melody, *args):
      # Only the first block has this source entrance. The preferred local
      # fill pairs with the next block's first attack, so it is ineligible.
      eligible = melody[0][0] == 0
      cost = (-2 if events[-1].start == fractions.Fraction(15, 2) else -1 if events[-1].start == 7 else 0) if eligible else 0
      return [('test_fill', cost)]
    # Distinguish source contexts without relying on the pattern or part ID.
    contexts = tuple(m if i < 2 else tuple((a + fractions.Fraction(1, 4), b) for a, b in m) for i, m in enumerate(melody))
    context = generator.comping_rhythm.acoustic_context(source_groups, contexts)
    with generator.vocabulary.using(resources, 'acoustic_accompaniment'), unittest.mock.patch('generator.comping_rhythm.evaluate', side_effect=evaluated):
      generator.comping_rhythm.add_fills(source_groups, selected, locations, 'acoustic', contexts, 'fill', context=context)
    self.assertEqual(locations[0]['fill']['pattern'], 'base.fill.safe')
    final = generator.comping_rhythm.realize_acoustic_form([r for _, _, r in selected], context)
    self.assertEqual([r.events for r in original[1:]], [r.events for r in final[1:]])
  def test_melody_and_progression_influence_selection_with_fixed_seed(self):
    source = tests.solo_fixtures.song(count=16, chords=('C7', 'FM7'))
    sparse = copy.deepcopy(source)
    for m in sparse['staves'][0]['measures']:
      m['voices'][0]['notes'] = [generator.model.make_event(m['id'] + '_held', fractions.Fraction(), fractions.Fraction(1, 2), 'note', [{'letter': 'c', 'alter': 0, 'octave': 4}])]
    choices = [generator.comping_rhythm.choose_cycles(groups(s), 'electric', generator.comping_rhythm.melody_contexts(s), 'fixed')[2] for s in (source, sparse)]
    self.assertNotEqual(choices[0][0]['patterns'], choices[1][0]['patterns'])
    changed = copy.deepcopy(source)
    for m in changed['staves'][0]['measures']:
      m['harmonies'].append({**generator.harmony.parse_symbol('G7'), 'id': 'change', 'onset': generator.model.rational(fractions.Fraction(3, 2))})
    other = generator.comping_rhythm.choose_cycles(groups(changed), 'electric', generator.comping_rhythm.melody_contexts(changed), 'fixed')[2]
    self.assertNotEqual(choices[0][0]['cost'], other[0]['cost'])
  def test_sparse_fill_replaces_only_second_bar_and_does_not_change_phase(self):
    source = tests.solo_fixtures.song(count=32, chords=('C7',))
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['electric']
    policy['groove']['cycle_patterns'] = 1
    policy['patterns'] = [p for p in policy['patterns'] if p['id'] in ('electric.charleston.statement', 'electric.charleston.answer')]
    # Section-ending evidence is source data. Deliberately put it at nearby
    # positions to exercise spacing, including the cyclic end/start join.
    for i in (2, 4, 12, 20, 30):
      source['staves'][0]['measures'][i]['section_label'] = 'Source section'
    realize = generator.comping_rhythm.realize
    def evaluated(pattern, inputs, *args, **kwargs):
      candidate = realize(pattern, inputs, *args, **kwargs)
      cost = -1 if '.fill.' in pattern['id'] and pattern['id'].endswith('.answer') else 0 if pattern['id'].endswith('.statement') else 1
      return dataclasses.replace(candidate, contributions=(('test_context', cost),))
    with generator.vocabulary.using(resources, 'electric_accompaniment'), unittest.mock.patch('generator.comping_rhythm.realize', side_effect=evaluated):
      selected, locations, _ = generator.comping_rhythm.choose_cycles(groups(source), 'electric', generator.comping_rhythm.melody_contexts(source), 'fills')
    fills = [v['fill']['source_index'] for v in locations if v['fill']]
    self.assertGreater(len(fills), 0)
    self.assertLessEqual(len(fills), 4)
    for i, a in enumerate(fills):
      for b in fills[i + 1:]:
        self.assertGreaterEqual(min(abs(a - b), 32 - abs(a - b)), 8)
    original = policy['patterns'][0]
    for pattern, _, _ in selected:
      self.assertEqual([r for r in pattern['events'] if fractions.Fraction(r['onset']) < 4], [r for r in original['events'] if fractions.Fraction(r['onset']) < 4])
    self.assertTrue(all(v['cycle_slot'] == 0 for v in locations))
  def test_real_source_fit_can_select_a_fill_without_mocked_scores(self):
    source = tests.solo_fixtures.song(count=16, chords=('C7',))
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['electric']
    policy['groove']['cycle_patterns'] = 1
    policy['patterns'] = [{'id': 'test.' + name, 'family': 'call_response', 'lengths': ['4', '4'], 'events': [
      {'onset': '1/2', 'duration': '1/2', 'layer': 'electric'}, {'onset': onset, 'duration': '1/2', 'layer': 'electric'}]}
      for name, onset in (('base', '11/2'), ('answer', '5'))]
    # Most of the form favors the offbeat answer. At the final boundary the
    # source enters on beat two after a breath, so the related on-beat answer
    # supports that entrance. All contributions come from the real evaluator.
    source['staves'][0]['measures'][-1]['voices'][0]['notes'].pop(0)
    with generator.vocabulary.using(resources, 'electric_accompaniment'):
      _, locations, reports = generator.comping_rhythm.choose_cycles(groups(source), 'electric', generator.comping_rhythm.melody_contexts(source), 'real-fill')
    self.assertEqual(reports[0]['patterns'], ['test.base'])
    self.assertEqual([r['fill']['source_index'] for r in locations if r['fill']], [15])
  def test_meter_change_restarts_compatible_cycle_and_reports_reduction(self):
    source = tests.solo_fixtures.song(count=8, chords=('C7',))
    shorter = tests.solo_fixtures.song(meter=3, count=4, chords=('C7',))
    source['staves'][0]['measures'][4:] = shorter['staves'][0]['measures']
    canonical = generator.comping.Canonical(source, 'electric', 'meter')
    self.assertEqual([r['meter'] for r in canonical.groove_report], ['4/4', '3/4'])
    self.assertTrue(all(r['reduction'] for r in canonical.groove_report))
    self.assertEqual(canonical.locations[4]['cycle_slot'], 0)
  def test_partial_entry_uses_second_half_instead_of_restarting(self):
    source = tests.solo_fixtures.song(count=10, chords=('C7',))
    canonical = generator.comping.Canonical(source, 'electric', 'entry')
    measures = source['staves'][0]['measures']
    full = canonical(measures[3], measures[2]['harmonies'][0], 3)
    sliced = canonical(measures[3], measures[2]['harmonies'][0], 3, bounds=(3, 10))
    # The entry may attack a formerly carried chord; actual relative times
    # and fret positions remain from the same second half of the figure.
    def music(m):
      return [(generator.model.fraction(n['onset']), generator.model.fraction(n['duration']), [(generator.model.pitch_number(p), p['tablature']) for p in n['pitches']]) for n in generator.score.measure_notes(m)]
    self.assertEqual(music(full), music(sliced))
  def test_final_harmony_change_keeps_unaffected_rhythm_and_grips(self):
    source = tests.solo_fixtures.song(count=8, chords=('C7',))
    canonical = generator.comping.Canonical(source, 'acoustic', 'final')
    original = source['staves'][0]['measures']
    changed = copy.deepcopy(original)
    changed[-1]['harmonies'] = [{**generator.harmony.parse_symbol('FM7'), 'id': 'final', 'onset': generator.model.rational(0)}]
    before = canonical(original[6], original[5]['harmonies'][0], 6)
    after = canonical(changed[6], changed[5]['harmonies'][0], 6, sources=changed)
    self.assertEqual(signature(before), signature(after))
    self.assertEqual(canonical.rhythm_report[-1]['base_pattern'], canonical.locations[6]['base_pattern'])
  def test_bass_eighth_then_upper_quarter_survives_single_voice_merge(self):
    source = tests.solo_fixtures.song(count=2, chords=('Dm7',))
    source['staves'][0]['measures'][0]['harmonies'].append({**generator.harmony.parse_symbol('G7'), 'id': 'middle', 'onset': generator.model.rational(2)})
    resources = copy.deepcopy(generator.vocabulary.load())
    policy = resources['accompaniment']['instruments']['acoustic']
    policy['patterns'] = [p for p in policy['patterns'] if p['id'] == 'acoustic.quarters.overlapping']
    with generator.vocabulary.using(resources, 'acoustic_accompaniment'):
      canonical = generator.comping.Canonical(source, 'acoustic', 'overlap')
      source['staves'][0]['measures'] = [canonical(m, None, i) for i, m in enumerate(source['staves'][0]['measures'])]
    source['staves'][0].update(arrangement_part_type='acoustic', tablature=True)
    generator.acoustic.validate_comping(source)
    generator.acoustic.combine(source)
    generator.validation.validate_document(source)
    generator.fretboard.validate_document(source)
    generator.acoustic.validate_comping(source)
    first = source['staves'][0]['measures'][0]
    self.assertEqual(len(first['voices']), 1)
    notes = [n for n in generator.score.measure_notes(first) if n['pitches']]
    self.assertEqual([(generator.model.fraction(n['onset']), generator.model.fraction(n['duration'])) for n in notes],
      [(0, fractions.Fraction(1, 2)), (fractions.Fraction(1, 2), 1), (2, 1)])
    strings = [{p['tablature']['string'] for p in n['pitches']} for n in notes]
    self.assertIn(strings[0], ({5}, {6}))
    self.assertEqual(strings[1], {2, 3, 4})
    self.assertIn(strings[2], ({2, 3, 4, 5}, {2, 3, 4, 6}))
    self.assertFalse(any(p.get('tie_from') or p.get('tie_to') for n in notes for p in n['pitches']))
  def test_mid_measure_chord_change_bass_without_forced_barline_attack(self):
    source = tests.solo_fixtures.song(count=2, chords=('C7',))
    m = source['staves'][0]['measures'][0]
    m['harmonies'].append({**generator.harmony.parse_symbol('FM7'), 'id': 'middle', 'onset': generator.model.rational(2)})
    pattern = {'id': 'test', 'family': 'quarters', 'lengths': ['4'], 'events': [
      {'onset': '1/2', 'duration': '1', 'layer': 'upper'}, {'onset': '5/2', 'duration': '1/2', 'layer': 'upper'}]}
    events, _ = generator.comping_rhythm.fit(pattern, m, None, 'acoustic')
    self.assertEqual([(e.start, e.lane) for e in events], [(fractions.Fraction(1, 2), 'combined'), (fractions.Fraction(5, 2), 'combined')])
    # Unpaired attacks include bass even when the harmony is unchanged.
    inputs = ((source['staves'][0]['measures'][1], source['staves'][0]['measures'][1]['harmonies'][0], 1),)
    r = generator.comping_rhythm.realize(pattern, inputs, 'acoustic', generator.comping_rhythm.melody_contexts(source))
    self.assertTrue(all(e.lane == 'combined' for e in r.events))
  def test_clean_cached_parallel_and_other_role_settings_keep_groove(self):
    source = generator.source.SourceAnalysis.read(tests.solo_fixtures.song(count=10, chords=('C7', 'FM7')))
    root = pathlib.Path('data').resolve()
    roles = ('acoustic_accompaniment', 'electric_accompaniment')
    clean = {r: generator.composition.generate_role(source, r, root) for r in roles}
    with tempfile.TemporaryDirectory() as directory:
      for r in roles:
        cold = generator.composition.generate_role(source, r, root, cache_dir=directory)
        warm = generator.composition.generate_role(source, r, root, cache_dir=directory)
        self.assertTrue(warm['cache_hit'])
        self.assertEqual(clean[r]['document'], cold['document'])
        self.assertEqual(clean[r]['document'], warm['document'])
      generator.composition.generate_role(source, roles[1], root, {'seed': 'different'}, directory)
      self.assertEqual(clean[roles[0]]['document'], generator.composition.generate_role(source, roles[0], root, cache_dir=directory)['document'])
    with concurrent.futures.ProcessPoolExecutor(max_workers=2, mp_context=multiprocessing.get_context('spawn')) as pool:
      futures = {r: pool.submit(generator.production._job, generator.composition_types.RoleRequest(source, r, root)) for r in reversed(roles)}
      for r, future in futures.items():
        self.assertEqual(clean[r]['document'], future.result()['document'])
  def test_reuse_audit_checks_rests_and_all_six_roles(self):
    source = tests.solo_fixtures.song(count=2, chords=('C7',))
    for role in generator.composition_types.ROLES:
      part = role.split('_')[0]
      wanted = 'slash' if role.endswith('_solo') else 'vocal' if part == 'vocal' else 'melody' if role.endswith('_melody') else 'accompaniment'
      section = generator.arrange.form.ArrangementSection('Repeated role', 0, 2, 1, wanted)
      doc = copy.deepcopy(source)
      doc['staves'][0]['measures'] *= 2
      doc = copy.deepcopy(doc)
      with unittest.mock.patch('generator.arrange.form.full_arrangement_sections', return_value=[section, section]):
        result = generator.arrange.accompaniment.audit(source, {role: doc}, role_documents=True)
        self.assertEqual(result[role]['reused_bars'], 2)
        # Remove music only in the second occurrence; reuse must fail.
        doc['staves'][0]['measures'][2] = copy.deepcopy(doc['staves'][0]['measures'][2])
        n = doc['staves'][0]['measures'][2]['voices'][0]['notes'][0]
        n.update(kind='rest', pitches=[])
        with self.assertRaisesRegex(ValueError, 'canonical'):
          generator.arrange.accompaniment.audit(source, {role: doc}, role_documents=True)

if __name__ == '__main__':
  unittest.main()