"""Dependency and music contracts use synthetic sources, never audition scores."""
import concurrent.futures
import copy
import contextlib
import dataclasses
import json
import multiprocessing
import pathlib
import tempfile
import unittest
import fractions
import unittest.mock

import generator.production
import generator.model
import generator.vocabulary
import generator.solo.expressions
import generator.solo.types
import generator.solo.harmonic
import generator.solo.placement
import generator.solo.notation
import generator.solo.composer
import tests.solo_fixtures
import tests.solo_fixtures
import tests.solo_fixtures

def music(result):
  return result['document']

class IndependentRoles(unittest.TestCase):
  @classmethod
  def setUpClass(cls):
    cls.source = generator.production.SourceAnalysis.read(tests.solo_fixtures.song(count=1, chords=('CM7',)))
    cls.root = pathlib.Path('data').resolve()
  def test_alone_order_parallel_and_assembly(self):
    serial = {role: generator.production.generate_role(self.source, role, self.root) for role in generator.production.ROLES}
    with concurrent.futures.ProcessPoolExecutor(max_workers=2, mp_context=multiprocessing.get_context('spawn')) as pool:
      jobs = {role: pool.submit(generator.production._job, (self.source, role, self.root, {}, None)) for role in reversed(generator.production.ROLES)}
      parallel = {role: future.result() for role, future in jobs.items()}
    for role in generator.production.ROLES:
      self.assertEqual(music(serial[role]), music(parallel[role]), role)
    for doc in generator.production.assemble(self.source, serial).values():
      generator.model.validate_document(doc)
  def test_role_setting_and_cache_are_independent(self):
    with tempfile.TemporaryDirectory() as directory:
      a = generator.production.generate_role(self.source, 'acoustic_solo', self.root, cache_dir=directory)
      b = generator.production.generate_role(self.source, 'electric_solo', self.root, {'seed': 'changed'}, directory)
      again = generator.production.generate_role(self.source, 'acoustic_solo', self.root, cache_dir=directory)
      clean = generator.production.generate_role(self.source, 'acoustic_solo', self.root)
      self.assertTrue(again['cache_hit'])
      self.assertEqual(music(a), music(again))
      self.assertEqual(music(a), music(clean))
      self.assertEqual(json.dumps(music(a)), json.dumps(music(again)))
      self.assertNotEqual(a['trace']['cache_key'], b['trace']['cache_key'])
  def test_optional_inputs_cannot_be_opened_by_composition(self):
    original = pathlib.Path.open
    def guarded(path, *args, **kwargs):
      self.assertNotIn('audition', str(path))
      self.assertNotIn('reference', str(path))
      self.assertNotIn('generated_assets', str(path))
      return original(path, *args, **kwargs)
    with unittest.mock.patch.object(pathlib.Path, 'open', guarded):
      result = generator.production.generate_role(self.source, 'acoustic_solo', self.root)
    self.assertNotIn('reference', json.dumps(result['trace']))
  def test_titles_and_export_ids_do_not_supply_musical_identity(self):
    document = tests.solo_fixtures.song(count=1, chords=('CM7',))
    other = copy.deepcopy(document)
    other['metadata'].update(title='A completely different title', composer='Different')
    for m in other['staves'][0]['measures']:
      m['id'] = 'exporter_123'
      m['number'] = '99'
    self.assertEqual(generator.production.SourceAnalysis.read(document), generator.production.SourceAnalysis.read(other))
  def test_reference_presence_and_changes_do_not_change_clean_music(self):
    # No repository reference is a fixture or prerequisite. These deliberately
    # invalid optional files test the absence of a discovery/fallback branch.
    with tempfile.TemporaryDirectory() as directory, contextlib.chdir(directory):
      reference_dir = pathlib.Path(directory) / 'build'
      reference_dir.mkdir()
      references = [reference_dir / (source_name + '-solo-' + str(index) + suffix + '.xml')
        for source_name in ('all-of-me', 'autumn-leaves') for index in (1, 2) for suffix in ('', '(4)')]
      before = generator.production.generate_role(self.source, 'electric_solo', self.root)
      for reference in references:
        reference.write_text('<not-a-production-source/>')
      present = generator.production.generate_role(self.source, 'electric_solo', self.root)
      for reference in references:
        reference.write_text('changed optional target, still not MusicXML')
      changed = generator.production.generate_role(self.source, 'electric_solo', self.root)
      for reference in references:
        reference.unlink()
      absent = generator.production.generate_role(self.source, 'electric_solo', self.root)
      self.assertTrue(all(music(before) == music(v) for v in (present, changed, absent)))
  def test_original_melody_changes_the_theme_with_unchanged_harmony(self):
    source = tests.solo_fixtures.song(count=4, chords=('CM7',))
    changed = copy.deepcopy(source)
    for m in changed['staves'][0]['measures']:
      for v in m['voices']:
        for n in v['notes']:
          for p in n['pitches']:
            p['letter'] = {'c': 'e', 'd': 'f', 'e': 'g', 'f': 'a'}[p['letter']]
    a = generator.production.generate_role(generator.production.SourceAnalysis.read(source), 'acoustic_solo', self.root)
    b = generator.production.generate_role(generator.production.SourceAnalysis.read(changed), 'acoustic_solo', self.root)
    self.assertNotEqual(a['details']['phrases'][0]['notes'], b['details']['phrases'][0]['notes'])
  def test_source_tuplet_lanes_survive_assembly(self):
    serial = {r: generator.production.generate_role(self.source, r, self.root) for r in generator.production.ROLES}
    # Exercise a different original lane name, as in the imported Autumn head.
    doc = serial['electric_melody']['document']
    staff = doc['staves'][0]
    raw = tuple(tests.solo_fixtures.note(fractions.Fraction(i, 3), fractions.Fraction(i + 1, 3), 'c', event_id=str(i)) for i in range(3))
    placed = generator.solo.placement.place(raw, fractions.Fraction(120))[1]
    index = next(i for i, m in enumerate(staff['measures']) if m.get('section_label') == 'Guitar Melody')
    replacement = generator.solo.notation.measures(staff['measures'][index:index + 1], placed, 'source_triplet')
    generator.solo.notation.replace_region(doc, index, replacement)
    for m in staff['measures']:
      for v in m['voices']:
        v['id'] = 'voice_source'
        for n in v['notes']:
          n['voice'] = 'voice_source'
    for t in staff['tuplets']:
      t['voice'] = 'voice_source'
    generator.model.validate_document(doc)
    generator.model.validate_document(generator.production.assemble(self.source, serial)['electric'])
  def test_sparse_parts_reuse_the_canonical_form(self):
    result = generator.production.generate_role(generator.production.SourceAnalysis.read(tests.solo_fixtures.song(count=4, chords=('Dm7', 'G7', 'CM7'))), 'acoustic_accompaniment', self.root)
    measures = result['document']['staves'][0]['measures']
    def signature(m):
      return [(n['onset'], n['duration'], [(generator.model.pitch_number(p), p['tablature']) for p in n['pitches']]) for v in m['voices'] for n in v['notes']]
    self.assertEqual([signature(m) for m in measures[4:8]], [signature(m) for m in measures[8:12]])
    self.assertTrue(all(len(n['pitches']) in (0, 1, 3, 4) for m in measures for v in m['voices'] for n in v['notes']))

class TypedExpressionContracts(unittest.TestCase):
  def setUp(self):
    self.h = generator.solo.harmonic.Harmony(tests.solo_fixtures.choices(('C7', 'FM7')))
  def make(self, rule='upper_lower'):
    target = tests.solo_fixtures.note(2, 4, 'e', event_id='target')
    upper = tests.solo_fixtures.note(1, fractions.Fraction(3, 2), 'f', event_id='upper')
    lower = dataclasses.replace(tests.solo_fixtures.note(fractions.Fraction(3, 2), 2, 'd', event_id='lower'), pitch=generator.solo.types.Pitch('d', fractions.Fraction(1), 4))
    c = generator.solo.types.ExpressionContract('gesture', 'enclosure', rule, (upper, lower), target, 'C7')
    return (upper, lower, target), c
  def test_complete_enclosure_and_orphans(self):
    notes, c = self.make()
    self.assertTrue(self.h.evaluate(notes, expressions=(c,)).valid)
    self.assertTrue(self.h.evaluate(notes).valid)
    self.assertFalse(self.h.evaluate(notes, infer=False).valid)
    self.assertFalse(self.h.evaluate(notes[:-1], expressions=(c,)).valid)
    changed = (*notes[:-1], dataclasses.replace(notes[-1], pitch=dataclasses.replace(notes[-1].pitch, octave=5)))
    self.assertFalse(self.h.evaluate(changed, expressions=(c,)).valid)
  def test_anticipation_validates_actual_future_harmony(self):
    n = tests.solo_fixtures.note(fractions.Fraction(7, 2), 6, 'a', event_id='early')
    c = generator.solo.types.ExpressionContract('anticipation', 'anticipation', 'early_chord_tone', (n,), n, 'FM7', boundary=fractions.Fraction(4))
    self.assertTrue(self.h.evaluate((n,), expressions=(c,)).valid)
    self.assertTrue(self.h.evaluate((n,)).valid)
    self.assertFalse(self.h.evaluate((n,), infer=False).valid)
    self.assertFalse(self.h.evaluate((n,), expressions=(dataclasses.replace(c, boundary=fractions.Fraction(3)),)).valid)
  def test_optional_color_is_not_a_written_chord_requirement(self):
    n = tests.solo_fixtures.note(0, 2, 'd', event_id='ninth')
    c = generator.solo.types.ExpressionContract('color', 'color', 'dominant_ninth', (n,), n, 'C7')
    self.assertTrue(self.h.evaluate((n,), expressions=(c,)).valid)
    self.assertTrue(self.h.evaluate((n,)).valid)
    self.assertFalse(self.h.evaluate((n,), infer=False).valid)
    self.assertNotIn(n.pitch.number() % 12, self.h.spans[0].roles.basic)
    self.assertNotIn(n.pitch.number() % 12, self.h.spans[0].roles.specified)
  def test_labels_do_not_authorize_unresolved_notes(self):
    n = dataclasses.replace(tests.solo_fixtures.note(0, 2, 'f', event_id='bad'), pitch=generator.solo.types.Pitch('f', fractions.Fraction(1), 4), role='color')
    self.assertFalse(self.h.evaluate((n,)).valid)
    c = generator.solo.types.ExpressionContract('bad', 'color', 'anything', (n,), n, 'C7')
    self.assertFalse(self.h.evaluate((n,), expressions=(c,)).valid)
  def test_delayed_approach_allows_only_the_declared_rest(self):
    lower = dataclasses.replace(tests.solo_fixtures.note(1, fractions.Fraction(3, 2), 'd', event_id='lower'), pitch=generator.solo.types.Pitch('d', fractions.Fraction(1), 4))
    target = tests.solo_fixtures.note(2, 4, 'e', event_id='target')
    c = generator.solo.types.ExpressionContract('delay', 'approach', 'delayed_below', (lower,), target, 'C7')
    self.assertTrue(self.h.evaluate((lower, target), expressions=(c,)).valid)
    wrong = dataclasses.replace(c, rule='below')
    self.assertFalse(self.h.evaluate((lower, target), expressions=(wrong,)).valid)
  def test_altered_dominant_color_requires_function_and_resolution(self):
    color = dataclasses.replace(tests.solo_fixtures.note(0, 1, 'd', event_id='color'), pitch=generator.solo.types.Pitch('d', fractions.Fraction(-1), 4))
    target = tests.solo_fixtures.note(1, 2, 'c', event_id='target')
    c = generator.solo.types.ExpressionContract('flat9', 'color', 'dominant_flat_ninth', (color,), target, 'C7')
    self.assertTrue(self.h.evaluate((color, target), expressions=(c,)).valid)
    unresolved = generator.solo.harmonic.Harmony(tests.solo_fixtures.choices(('C7', 'Dm7')))
    self.assertFalse(unresolved.evaluate((color, target), expressions=(c,)).valid)
    self.assertFalse(self.h.evaluate((color,), expressions=(c,)).valid)
  def test_mapped_semitone_needs_no_expression_exception(self):
    h = generator.solo.harmonic.Harmony(tests.solo_fixtures.choices(('A7', 'Dm7')))
    a = dataclasses.replace(tests.solo_fixtures.note(0, fractions.Fraction(1, 2), 'c', event_id='csharp'), pitch=generator.solo.types.Pitch('c', fractions.Fraction(1), 4))
    b = tests.solo_fixtures.note(fractions.Fraction(1, 2), 1, 'd', event_id='d')
    self.assertTrue(h.evaluate((a, b)).valid)
  def test_ties_and_barlines_do_not_restart_triplets(self):
    raw = tuple(tests.solo_fixtures.note(fractions.Fraction(7, 2) + fractions.Fraction(i, 3), fractions.Fraction(7, 2) + fractions.Fraction(i + 1, 3), 'c', event_id=str(i)) for i in range(3))
    placed = generator.solo.placement.place(raw, fractions.Fraction(120))[1]
    doc = tests.solo_fixtures.song(count=2, chords=('CM7',))
    measures = generator.solo.notation.measures(doc['staves'][0]['measures'], placed, 'triplet')
    spans = generator.solo.notation.tuplets(measures)
    self.assertEqual(len(spans), 1)
    self.assertTrue(any('tie_from' in p for m in measures for v in m['voices'] for n in v['notes'] for p in n['pitches']))

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