product-vamp/generator/solo/harmonic.py

"""Chord function on exact sounding intervals, separate from scale permission."""
from __future__ import annotations

import dataclasses
import fractions
import functools

import generator.harmony
import generator.model
import generator.solo.policy
import generator.solo.types

@dataclasses.dataclass(frozen=True)
class Roles:
  basic: frozenset[int]
  specified: frozenset[int]
  degrees: tuple[tuple[int, tuple[int, ...]], ...]

@functools.lru_cache(maxsize=512)
def degree_roles(kind, root, degrees):
  basic, specified, labels = set(), set(), {}
  for degree, alteration in degrees:
    pitch = int(root + generator.harmony.degree_semitones(degree, alteration)) % 12
    structural = degree in (1, 3, 5, 6, 7) or degree in (2, 4) and kind in ('suspended-second', 'suspended-fourth')
    (basic if structural else specified).add(pitch)
    labels.setdefault(pitch, []).append(degree)
  return Roles(frozenset(basic), frozenset(specified - basic), tuple((p, tuple(v)) for p, v in sorted(labels.items())))

def roles(chord):
  if not chord or chord['kind'] == 'none':
    return Roles(frozenset(), frozenset(), ())
  root = chord['root']
  number = generator.model.NATURAL[root['letter']] + generator.model.alteration(root['alter'])
  return degree_roles(chord['kind'], number, tuple(sorted(generator.harmony.degrees(chord).items())))

@dataclasses.dataclass(frozen=True)
class Span:
  start: fractions.Fraction
  end: fractions.Fraction
  roles: Roles
  scale: object
  chord: str

@dataclasses.dataclass(frozen=True)
class Result:
  valid: bool
  exposure: float
  pending: tuple[generator.solo.types.HarmonicTreatment, ...] = ()
  error: str = ''

class Harmony:
  def __init__(self, choices):
    self.choices = choices
    self.spans = tuple(Span(c.span.start, c.span.end, roles(c.span.chord), c.scale, generator.harmony.symbol(c.span.chord) if c.span.chord else 'N.C.') for c in choices)
  def at(self, time):
    return next((s for s in self.spans if s.start <= time < s.end), None)
  def segments(self, note):
    return tuple((s, max(note.start, s.start), min(note.end, s.end)) for s in self.spans if s.start < note.end and s.end > note.start)
  def members(self, start, end):
    spans = [s for s in self.spans if s.start < end and s.end > start]
    return frozenset.intersection(*(s.roles.basic for s in spans)) if spans else frozenset()
  def function(self, number, span):
    pitch = number % 12
    return 'basic' if pitch in span.roles.basic else 'specified_extension' if pitch in span.roles.specified else 'mapped_motion' if span.scale and pitch in span.scale.pitches else 'outside_collection'
  def step(self, first, second, time):
    span = self.at(time)
    if not span or not span.scale:
      return False
    try:
      return abs(generator.solo.types.scale_index(span.scale, first) - generator.solo.types.scale_index(span.scale, second)) == 1
    except ValueError:
      return False
  def connected_motion(self, notes, index):
    if not 0 < index < len(notes) - 1:
      return ''
    before, note, after = notes[index - 1:index + 2]
    if before.end != note.start or note.end != after.start:
      return ''
    a, b, c = (n.pitch.number() for n in (before, note, after))
    if self.step(a, b, note.start) and self.step(b, c, note.start):
      return 'neighbor' if a == c else 'passing' if (b - a) * (c - b) > 0 else ''
    return ''
  def evaluate(self, notes, treatments=(), *, until=None, complete=False):
    notes = generator.solo.types.sounding_notes(notes)
    until = until if until is not None else max((n.end for n in notes), default=fractions.Fraction())
    pending, valid_contracts = [], []
    for treatment in treatments:
      if treatment.start >= until:
        continue
      if treatment.kind == 'color':
        spans = [s for s in self.spans if s.start < treatment.end and s.end > treatment.start]
        if not spans or any(treatment.pitch % 12 not in s.roles.basic | s.roles.specified for s in spans):
          return Result(False, 0, error=f'{treatment.id}: color is not specified by the overlapped harmony')
        valid_contracts.append(treatment)
        continue
      if treatment.kind != 'suspension' or treatment.resolution is None or treatment.deadline is None:
        return Result(False, 0, error=f'{treatment.id}: incomplete harmonic treatment')
      held = next((n for n in notes if n.pitch.number() == treatment.pitch and n.start <= treatment.start - treatment.preparation and n.end > treatment.start), None)
      if held is None:
        return Result(False, 0, error=f'{treatment.id}: suspension lacks sounding preparation')
      if treatment.pitch % 12 not in self.members(treatment.start - treatment.preparation, treatment.start):
        return Result(False, 0, error=f'{treatment.id}: preparation is not a chord member')
      resolution = next((n for n in notes if n.start == held.end and n.pitch.number() == treatment.resolution and n.start <= treatment.deadline), None)
      if resolution:
        span = self.at(resolution.start)
        if not span or treatment.resolution % 12 not in span.roles.basic or not self.step(treatment.pitch, treatment.resolution, resolution.start):
          return Result(False, 0, error=f'{treatment.id}: resolution is not an adjacent chord member')
      elif complete or until > treatment.deadline or any(n.start >= held.end for n in notes if n is not held):
        return Result(False, 0, error=f'{treatment.id}: unresolved suspension at deadline {treatment.deadline}')
      else:
        pending.append(treatment)
      valid_contracts.append(treatment)
    exposed, sounding = 0.0, 0.0
    for index, note in enumerate(notes):
      number = note.pitch.number()
      duration = note.end - note.start
      sounding += float(duration)
      covered = note.start
      for span, start, end in self.segments(note):
        if start > covered or not span.scale or number % 12 not in span.scale.pitches:
          return Result(False, 0, error=f'{note.pitch.label()} at {start}-{end}: outside mapped collection or missing harmony')
        covered = end
        function = self.function(number, span)
        if function == 'basic' or note.role == 'quote':
          continue
        contract = next((t for t in valid_contracts if t.pitch == number and t.start <= start and t.end >= end), None)
        if duration >= generator.solo.policy.DEFAULT.long_note and contract is None:
          return Result(False, 0, error=f'{note.pitch.label()} at {start}-{end} over {span.chord}: sustained {function} without a planned treatment')
        connected = duration < generator.solo.policy.DEFAULT.stopping_note and self.connected_motion(notes, index)
        cost = 0 if contract or connected else 1 if duration >= generator.solo.policy.DEFAULT.stopping_note else 0.25
        exposed += float(end - start) * cost
      if covered < note.end:
        return Result(False, 0, error=f'{note.pitch.label()} at {covered}: missing harmonic coverage')
    return Result(True, exposed / max(sounding, 0.001), tuple(pending))
  def describe(self, notes, treatments=()):
    notes = generator.solo.types.sounding_notes(notes)
    result = []
    for index, note in enumerate(notes):
      for span, start, end in self.segments(note):
        result.append({'start': str(start), 'end': str(end), 'attack': str(note.start), 'sounding_duration': str(note.end - note.start), 'pitch': note.pitch.label(), 'chord': span.chord, 'function': self.function(note.pitch.number(), span), 'motion': self.connected_motion(notes, index), 'protected': note.role == 'quote', 'treatments': [t.id for t in treatments if t.pitch == note.pitch.number() and t.start <= start and t.end >= end]})
    return result

def anchor_note(anchor, notes, *, until=None):
  for note in notes:
    if note.pitch.number() != anchor.pitch.number():
      continue
    arrival = note.start if anchor.onset_required else max(note.start, anchor.earliest)
    needed = max(arrival + anchor.duration, anchor.hold_until or arrival)
    if anchor.earliest <= arrival <= anchor.latest and note.start <= arrival < note.end and note.end >= min(needed, until if until is not None else needed):
      return note
  return None

def anchors_valid(anchors, notes, until, *, complete=False, closure='open'):
  for anchor in anchors:
    if anchor.earliest >= until and not complete:
      continue
    note = anchor_note(anchor, notes, until=None if complete else until)
    if note is None and (complete or until > anchor.latest):
      return False, f'{anchor.id}: required {anchor.role} arrival {anchor.earliest}-{anchor.latest} on {anchor.pitch.label()} not completed'
    if note and anchor.role == 'terminal' and closure == 'closed' and any(n.start > note.start for n in notes):
      return False, f'{anchor.id}: terminal anchor touched then abandoned'
  return True, ''