product-vamp/generator/solo/analysis.py

from __future__ import annotations

import dataclasses
import fractions
import hashlib
import json

import generator.harmony
import generator.model
import generator.score
import generator.solo.types

@dataclasses.dataclass(frozen=True)
class Analysis:
  motifs: tuple[generator.solo.types.Motif, ...]
  boundaries: tuple[fractions.Fraction, ...]
  tempo: fractions.Fraction
  melody: tuple[generator.solo.types.Note, ...]
  body_offset: fractions.Fraction = fractions.Fraction()
  source_phrases: tuple[generator.solo.types.SourcePhrase, ...] = ()

def identity(analysis, measures, part):
  anchor = analysis.melody[0].pitch.number() if analysis.melody else 0
  melody = [(str(n.start), str(n.end), n.pitch.number() - anchor) for n in analysis.melody]
  harmony = []
  for measure in measures:
    for chord in measure['harmonies']:
      pitches = sorted((int(p) - anchor) % 12 for p in generator.harmony.pitch_classes(chord)) if chord['kind'] != 'none' else []
      harmony.append((chord['kind'], pitches, chord['onset']))
  return hashlib.sha256(json.dumps((part, melody, harmony), sort_keys=True).encode()).hexdigest()

def attacks(document: dict) -> tuple[generator.solo.types.Note, ...]:
  """Collapse tied tones before extracting monophonic melody fragments."""
  result, pending = [], {}
  absolute = fractions.Fraction()
  for measure in generator.score.source_staff(document)['measures']:
    for event in generator.score.measure_notes(measure):
      if event.get('grace') or event.get('cue'):
        continue
      for tone in event['pitches']:
        number = generator.model.pitch_number(tone)
        if number.denominator != 1:
          continue
        start = absolute + generator.model.fraction(event['onset'])
        end = start + generator.model.fraction(tone['duration'])
        pitch = generator.solo.types.Pitch(tone['letter'], generator.model.alteration(tone['alter']), tone['octave'])
        source = pending.pop(tone.get('tie_from'), None)
        if source is not None:
          result[source] = dataclasses.replace(result[source], end=end, sources=(*result[source].sources, event['id']))
        else:
          source = len(result)
          result.append(generator.solo.types.Note(start, end, pitch, (event['id'],)))
        if 'tie_to' in tone:
          pending[tone['id']] = source
    absolute += generator.model.fraction(measure['duration'])
  return tuple(sorted(result, key=lambda n: (n.start, -n.pitch.number(), n.sources)))

def analyze(document: dict) -> Analysis:
  measures = generator.score.source_staff(document)['measures']
  melody = attacks(document)
  boundaries = [fractions.Fraction()]
  for measure in measures:
    boundaries.append(boundaries[-1] + generator.model.fraction(measure['duration']))
  tempo = next((generator.model.fraction(t['bpm']) * generator.model.written_length(t['written']) for m in measures for t in m.get('tempos', [])), fractions.Fraction(120))
  candidates = []
  for index, measure in enumerate(measures):
    start = boundaries[index]
    # Legacy motif consumers also receive complete longer source material.
    for count in (1, 2, 4, 6, 8):
      if index + count >= len(boundaries):
        continue
      end = boundaries[index + count]
      notes = tuple(n for n in melody if start <= n.start < end)
      if not notes or any(n.end > end for n in notes) or any(a.end > b.start for a, b in zip(notes, notes[1:])):
        continue
      normalized = tuple(dataclasses.replace(n, start=n.start - start, end=n.end - start) for n in notes)
      identity = repr(tuple((n.start, n.end, n.pitch) for n in normalized)).encode('ascii')
      identifier = 'head_' + hashlib.sha256(identity).hexdigest()[:12]
      context = tuple(generator.harmony.symbol(h) for m in measures[index:index + count] for h in m['harmonies'])
      candidates.append(generator.solo.types.Motif(identifier, normalized, end - start, context))
  unique = {m.id: m for m in reversed(candidates)}
  # Preserve the opening idea, then favor a contrasting rhythm/contour with space.
  ordered = list(unique.values())
  ordered.sort(key=lambda m: (next(i for i, n in enumerate(melody) if n.sources[0] == m.notes[0].sources[0]), m.duration))
  offset = boundaries[generator.score.body_start_measure_index(measures)]
  phrases = source_phrases(melody, offset, boundaries[-1] - offset,
    generator.model.fraction(measures[-1]['duration']))
  return Analysis(tuple(ordered), tuple(boundaries), tempo, melody, offset, phrases)

def source_phrases(melody, offset, end, bar):
  """Partition at musical releases, without slicing a sounding event at bars."""
  notes = tuple(dataclasses.replace(n, start=n.start-offset, end=n.end-offset,
    event_id=n.sources[0]) for n in melody if n.end > offset)
  if any(a.end > b.start for a, b in zip(notes, notes[1:])):
    raise ValueError('solo source requires a monophonic melody')
  candidates = sorted({fractions.Fraction(), end, *(n.end for n in notes if 0 < n.end <= end),
    *(n.start for n in notes if 0 < n.start < end)})
  candidates = [t for t in candidates if not any(n.start < t < n.end for n in notes)]
  if not candidates or candidates[0] != 0:
    candidates.insert(0, fractions.Fraction())
  best = {end: (0., ())}
  for start in reversed(candidates[:-1]):
    options = []
    for finish in candidates:
      duration = finish-start
      if finish not in best or duration <= 0 or duration > 8*bar or duration < 2*bar and finish != end:
        continue
      prior = next((n for n in reversed(notes) if n.end <= finish), None)
      after = next((n for n in notes if n.start >= finish), None)
      gap = float(max(0, after.start-prior.end)) if prior and after else 0.
      held = float(prior.end-prior.start) if prior else 0.
      cost = abs(float(duration/bar)-4)/4 + .7*(held < 2) - .3*min(2,gap)
      options.append((cost+best[finish][0], (finish,*best[finish][1])))
    if options:
      best[start] = min(options)
  if 0 not in best:
    # A single extended held event is one musical unit, regardless of length.
    stops = (end,)
  else:
    stops = best[0][1]
  result, start = [], fractions.Fraction()
  for finish in stops:
    actual = tuple(n for n in notes if n.start < finish and n.end > start)
    cursor, rests = start, []
    for n in actual:
      if cursor < n.start:
        rests.append((cursor,n.start))
      cursor = n.end
    if cursor < finish:
      rests.append((cursor,finish))
    origin = actual[0].pitch.number() if actual else 0
    shape = tuple((n.start-start,n.end-start,n.pitch.number()-origin) for n in actual)
    family = hashlib.sha256(repr(shape).encode()).hexdigest()[:12]
    pickup = tuple(dataclasses.replace(n,start=n.start-offset,end=n.end-offset,event_id=n.sources[0]) for n in melody if n.end<=offset) if start==0 else ()
    result.append(generator.solo.types.SourcePhrase(f'source_{start}_{finish}',start,finish,actual,tuple(rests),family,pickup))
    start = finish
  return tuple(result)