khodowany
208 lines
12 KiB
Python
208 lines
12 KiB
Python
"""Meter-specific vocabulary and source-derived quintuple grouping."""
|
|
import concurrent.futures
|
|
import copy
|
|
import fractions
|
|
import json
|
|
import multiprocessing
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
import generator.accompaniment
|
|
import generator.comping
|
|
import generator.comping_rhythm
|
|
import generator.model
|
|
import generator.production
|
|
import generator.vocabulary
|
|
import tests.solo_fixtures
|
|
import tests.test_comping_cycles
|
|
|
|
F = fractions.Fraction
|
|
|
|
class CompingMeters(unittest.TestCase):
|
|
def source(self, beats=5, count=10, explicit=()):
|
|
source = tests.solo_fixtures.song(meter=beats, count=count, chords=('C7', 'FM7'))
|
|
if explicit:
|
|
for m in source['staves'][0]['measures']:
|
|
m['meter']['groups'][0]['beats'] = list(explicit)
|
|
m['time_signature'] = generator.model.meter_label(m['meter'])
|
|
source['metadata']['time_signature'] = source['staves'][0]['measures'][0]['time_signature']
|
|
return source
|
|
def test_balanced_pools_and_all_families_reach_joint_evaluation(self):
|
|
for beats, grouping in ((4, ()), (3, ()), (5, (3, 2)), (5, (2, 3))):
|
|
source = self.source(beats, 2, grouping)
|
|
inputs = tests.test_comping_cycles.groups(source)[0]
|
|
flat = generator.comping_rhythm.flatten(inputs)
|
|
for role in ('acoustic', 'electric'):
|
|
with self.subTest(beats=beats, grouping=grouping, role=role):
|
|
policy = generator.vocabulary.accompaniment(role)
|
|
pool = generator.comping_rhythm.patterns_for(policy, flat, (F(beats),) * 2, grouping)
|
|
self.assertEqual(len(pool), 24)
|
|
self.assertEqual(len({generator.comping_rhythm.identity(p) for p in pool}), 24)
|
|
self.assertEqual(len({p['family'] for p in pool}), 11 if beats == 4 else 12)
|
|
candidates = generator.comping_rhythm.candidates(flat, None, role, (), 'coverage', lengths=(F(beats),) * 2, grouping=grouping)
|
|
self.assertEqual({p['family'] for p in pool}, {dict(c.families)['pattern'] for c in candidates})
|
|
for p in pool:
|
|
fitted = generator.comping_rhythm.realize(p, inputs, role, generator.comping_rhythm.melody_contexts(source))
|
|
self.assertIsNotNone(fitted, p['id'])
|
|
generator.comping_rhythm.validate_events(fitted.events, role)
|
|
for start in (0, beats):
|
|
self.assertLessEqual(len({e.start for e in fitted.events if start <= e.start < start + beats}), 3 if role == 'acoustic' else 2)
|
|
if role == 'electric':
|
|
self.assertTrue(any(F(e['onset']) < beats < F(e['onset']) + F(e['duration']) for p in pool for e in p['events']))
|
|
def test_explicit_grouping_overrides_harmony_and_melody(self):
|
|
source = self.source(explicit=(2, 3))
|
|
for m in source['staves'][0]['measures']:
|
|
m['harmonies'].append({**generator.harmony.parse_symbol('G7'), 'id': 'secondary', 'onset': generator.model.rational(3)})
|
|
grouping, evidence = generator.comping_rhythm.grouping_for(tests.test_comping_cycles.groups(source), generator.comping_rhythm.melody_contexts(source))
|
|
self.assertEqual(grouping, (2, 3))
|
|
self.assertEqual(evidence['reason'], 'written')
|
|
def test_harmonic_boundary_then_melodic_entrances_then_attacks(self):
|
|
for boundary in (2, 3):
|
|
source = self.source()
|
|
for m in source['staves'][0]['measures']:
|
|
m['harmonies'].append({**generator.harmony.parse_symbol('G7'), 'id': 'secondary', 'onset': generator.model.rational(boundary)})
|
|
grouping, _ = generator.comping_rhythm.grouping_for(tests.test_comping_cycles.groups(source), generator.comping_rhythm.melody_contexts(source))
|
|
self.assertEqual(grouping, (boundary, 5 - boundary))
|
|
source = self.source()
|
|
# Competing secondary pulses have equal attack counts. Only one follows
|
|
# an actual rest; an uninterrupted reattack is weaker grouping evidence.
|
|
other = 5 - boundary
|
|
contexts = tuple(((F(other) - F(1, 2), F(other)), (F(other), F(other) + F(1, 2)), (F(boundary), F(boundary) + F(1, 2))) for _ in range(10))
|
|
grouping, _ = generator.comping_rhythm.grouping_for(tests.test_comping_cycles.groups(source), contexts)
|
|
self.assertEqual(grouping, (boundary, 5 - boundary))
|
|
contexts = tuple(((F(boundary), F(boundary + 1)),) for _ in range(10))
|
|
grouping, _ = generator.comping_rhythm.grouping_for(tests.test_comping_cycles.groups(source), contexts)
|
|
self.assertEqual(grouping, (boundary, 5 - boundary))
|
|
def test_same_chord_marking_is_not_a_harmonic_change(self):
|
|
source = self.source()
|
|
for m in source['staves'][0]['measures']:
|
|
m['harmonies'].append({**m['harmonies'][0], 'onset': generator.model.rational(2)})
|
|
grouping, evidence = generator.comping_rhythm.grouping_for(tests.test_comping_cycles.groups(source), generator.comping_rhythm.melody_contexts(source))
|
|
self.assertEqual(grouping, (3, 2))
|
|
self.assertEqual(evidence['reason'], 'default_3+2')
|
|
def test_not_just_equal_measure_length(self):
|
|
source = self.source(3)
|
|
m = source['staves'][0]['measures'][0]
|
|
m.update(meter={'groups': [{'beats': [6], 'unit': 8}]}, time_signature='6/8')
|
|
with self.assertRaisesRegex(ValueError, 'no comping vocabulary for 6/8'):
|
|
generator.comping.Canonical(source, 'electric', 'compound')
|
|
def test_schema_rejects_wrong_meter_lengths_and_missing_grouping(self):
|
|
root = pathlib.Path('data/composition')
|
|
contents = {p.stem: p.read_text() for p in root.glob('*.json')}
|
|
for change in ({'meter': '3/4'}, {'grouping': []}, {'grouping': [4, 1]}):
|
|
value = json.loads(contents['accompaniment'])
|
|
pattern = next(p for p in value['instruments']['electric']['patterns'] if p['meter'] == '5/4')
|
|
pattern.update(change)
|
|
with self.assertRaises(ValueError):
|
|
generator.vocabulary._load(tuple((name, json.dumps(value) if name == 'accompaniment' else text) for name, text in contents.items()))
|
|
def test_full_cycles_odd_forms_and_grouping_consistency(self):
|
|
for beats, grouping in ((4, ()), (3, ()), (5, (3, 2)), (5, (2, 3))):
|
|
source = self.source(beats, 9, grouping)
|
|
reports = []
|
|
for role, width in (('acoustic', 2), ('electric', 4)):
|
|
canonical = generator.comping.Canonical(source, role, 'cycles')
|
|
report = canonical.groove_report[0]
|
|
reports.append(report)
|
|
self.assertEqual(report['cycle_patterns'], width)
|
|
self.assertEqual(report['eligible_patterns'], 24)
|
|
self.assertIsNone(report['reduction'])
|
|
for location in canonical.locations.values():
|
|
slot = location['cycle_slot']
|
|
self.assertEqual(location['base_pattern'], report['patterns'][slot])
|
|
last = canonical.block_rhythms[8]
|
|
self.assertTrue(all(0 <= e.start < e.end <= beats for e in last.events))
|
|
self.assertEqual(reports[0]['grouping'], reports[1]['grouping'])
|
|
def test_grouping_changes_restart_and_fills_cannot_cross_groups(self):
|
|
source = self.source(count=18, explicit=(3, 2))
|
|
for m in source['staves'][0]['measures'][9:]:
|
|
m['meter']['groups'][0]['beats'] = [2, 3]
|
|
m['time_signature'] = '2+3/4'
|
|
for role in ('acoustic', 'electric'):
|
|
canonical = generator.comping.Canonical(source, role, 'change')
|
|
self.assertEqual([r['grouping'] for r in canonical.groove_report], [[3, 2], [2, 3]])
|
|
self.assertEqual(canonical.locations[9]['cycle_slot'], 0)
|
|
for pattern, _ in canonical.patterns.values():
|
|
suffix = '.' + '+'.join(map(str, pattern['grouping'])) + '.'
|
|
if '.fill.' in pattern['id']:
|
|
self.assertTrue(all(suffix in p for p in pattern['id'].split('.fill.')))
|
|
def test_every_fill_candidate_uses_its_base_grouping(self):
|
|
source = self.source(count=16, explicit=(2, 3))
|
|
checked = []
|
|
realize = generator.comping_rhythm.realize
|
|
def inspect(pattern, *args, **kwargs):
|
|
if '.fill.' in pattern['id']:
|
|
checked.append(pattern['id'])
|
|
self.assertTrue(all('.2+3.' in p for p in pattern['id'].split('.fill.')))
|
|
return realize(pattern, *args, **kwargs)
|
|
with unittest.mock.patch.object(generator.comping_rhythm, 'realize', side_effect=inspect):
|
|
generator.comping.Canonical(source, 'electric', 'fill-group')
|
|
self.assertTrue(checked)
|
|
def test_partial_pickup_and_mid_pattern_entry_keep_meter_and_phase(self):
|
|
for beats in (3, 5):
|
|
source = self.source(beats)
|
|
m = copy.deepcopy(source['staves'][0]['measures'][0])
|
|
m['duration'] = generator.model.rational(F(3, 2))
|
|
candidates = generator.comping_rhythm.candidates(m, None, 'electric', (), 'pickup')
|
|
self.assertTrue(candidates)
|
|
self.assertTrue(all(0 <= e.start < e.end <= F(3, 2) for c in candidates for e in c.events))
|
|
for role in ('acoustic', 'electric'):
|
|
canonical = generator.comping.Canonical(source, role, 'entry')
|
|
measures = source['staves'][0]['measures']
|
|
carried = generator.accompaniment.carried_chords(measures)
|
|
full = canonical(measures[3], carried[3], 3)
|
|
partial = canonical(measures[3], carried[3], 3, bounds=(3, 10))
|
|
# Rest engraving can split differently at a sliced boundary. Sounded
|
|
# onsets, releases, voicings and tie relationships must remain equal.
|
|
self.assertEqual([n for n in tests.test_comping_cycles.signature(full) if n[-1]], [n for n in tests.test_comping_cycles.signature(partial) if n[-1]])
|
|
def test_unused_meter_and_other_role_data_do_not_reseed(self):
|
|
resources = generator.vocabulary.load()
|
|
measures = self.source(4)['staves'][0]['measures']
|
|
baseline = generator.production.dependencies('acoustic_accompaniment', resources, measures)
|
|
changed = copy.deepcopy(resources)
|
|
for role, policy in changed['accompaniment']['instruments'].items():
|
|
for p in policy['patterns']:
|
|
if role == 'electric' or p['meter'] in ('3/4', '5/4'):
|
|
p['events'] = []
|
|
self.assertEqual(baseline, generator.production.dependencies('acoustic_accompaniment', changed, measures))
|
|
changed['accompaniment']['instruments']['acoustic']['patterns'][0]['events'] = []
|
|
self.assertNotEqual(baseline, generator.production.dependencies('acoustic_accompaniment', changed, measures))
|
|
def test_chord_changes_anticipations_and_cross_bar_ties(self):
|
|
for beats, grouping in ((4, ()), (3, ()), (5, (3, 2)), (5, (2, 3))):
|
|
source = self.source(beats, 2, grouping)
|
|
for m in source['staves'][0]['measures']:
|
|
m['harmonies'].append({**generator.harmony.parse_symbol('G7'), 'id': 'middle', 'onset': generator.model.rational(beats - 1)})
|
|
inputs = tests.test_comping_cycles.groups(source)[0]
|
|
for role in ('acoustic', 'electric'):
|
|
pool = generator.comping_rhythm.patterns_for(generator.vocabulary.accompaniment(role), inputs[0][0], (F(beats),) * 2, grouping)
|
|
anticipated = False
|
|
for p in pool:
|
|
result = generator.comping_rhythm.realize(p, inputs, role, generator.comping_rhythm.melody_contexts(source), anticipate=True)
|
|
self.assertIsNotNone(result, p['id'])
|
|
generator.comping_rhythm.validate_events(result.events, role)
|
|
for e in result.events:
|
|
if e.harmonic_time is not None:
|
|
anticipated = True
|
|
self.assertEqual(e.harmonic_time - e.start, F(1, 2))
|
|
self.assertTrue(anticipated)
|
|
def test_roles_are_deterministic_alone_cached_and_parallel(self):
|
|
for meter in (3, 5):
|
|
source = generator.production.SourceAnalysis.read(self.source(meter, 9))
|
|
roles = ('acoustic_accompaniment', 'electric_accompaniment')
|
|
with tempfile.TemporaryDirectory() as cache:
|
|
clean = {r: generator.production.generate_role(source, r, pathlib.Path('data')) for r in roles}
|
|
for role in reversed(roles):
|
|
cold = generator.production.generate_role(source, role, pathlib.Path('data'), cache_dir=cache)
|
|
warm = generator.production.generate_role(source, role, pathlib.Path('data'), cache_dir=cache)
|
|
self.assertEqual(clean[role]['document'], cold['document'])
|
|
self.assertEqual(cold['document'], warm['document'])
|
|
self.assertTrue(warm['cache_hit'])
|
|
with concurrent.futures.ProcessPoolExecutor(max_workers=2, mp_context=multiprocessing.get_context('spawn')) as pool:
|
|
jobs = {r: pool.submit(generator.production._job, (source, r, pathlib.Path('data'), {}, None)) for r in reversed(roles)}
|
|
for role, job in jobs.items():
|
|
self.assertEqual(clean[role]['document'], job.result()['document'])
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|