"""Meter-specific vocabulary and source-derived quintuple grouping.""" import generator.composition import generator.harmony import generator.source 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) full_position = canonical.last_position partial = canonical(measures[3], carried[3], 3, bounds=(3, 10)) self.assertEqual(canonical.last_position, full_position) # 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.composition.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.composition.dependencies('acoustic_accompaniment', changed, measures)) changed['accompaniment']['instruments']['acoustic']['patterns'][0]['events'] = [] self.assertNotEqual(baseline, generator.composition.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.source.SourceAnalysis.read(self.source(meter, 9)) roles = ('acoustic_accompaniment', 'electric_accompaniment') with tempfile.TemporaryDirectory() as cache: clean = {r: generator.composition.generate_role(source, r, pathlib.Path('data')) for r in roles} for role in reversed(roles): cold = generator.composition.generate_role(source, role, pathlib.Path('data'), cache_dir=cache) warm = generator.composition.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, generator.composition_types.RoleRequest(source, r, pathlib.Path('data'))) for r in reversed(roles)} for role, job in jobs.items(): self.assertEqual(clean[role]['document'], job.result()['document']) if __name__ == '__main__': unittest.main()