fix and optimise the code, add new notebook

evaluation_function/compare_MIDI.py

@@ -17,6 +17,7 @@
 
 
 import numpy as np
+from collections import Counter
 
 # Default thresholds / parameters
 # Teachers can override any of these via the params dict in evaluation_function.
@@ -94,12 +95,14 @@ def identify_chord_name(notes):
 
 def compute_chord_accuracy(ref_notes, res_notes):
     """
-    Compute the chord accuracy score A from (Devaney, n.d.): 
+    Compute the chord accuracy score A (modified Devaney's method), but
+    use the actual MIDI pitch (with octave) instead of pitch class, 
+    so that the score is sensitive to octave errors.: 
     A = (C - I + |y|) / (2 * |y|)
     where:
-        C = |y ∩ y_hat|  (correctly played pitch classes)
-        I = |y_hat - y|  (unexpected pitch classes played)
-        |y|              (number of pitch classes in the reference chord)
+        C = number of correctly matched notes
+        I = number of extra notes
+        |y| = number of notes in the reference chord
     A = 1.0 means perfectly correct. 
     A = 0.0 means nothing correct and many unexpected notes are played.
 
@@ -109,28 +112,43 @@ def compute_chord_accuracy(ref_notes, res_notes):
 
     Returns:
         accuracy: float in [0, 1]
-        correct_pitches: sorted list of pitch class ints in both chords
-        missing_pitches: sorted list of pitch class ints in ref
-        extra_pitches: sorted list of pitch class ints in response
+        correct_pitches: sorted list of matched MIDI pitches
+        missing_pitches: sorted list of missing MIDI pitches
+        extra_pitches: sorted list of extra MIDI pitches
     """
-    ref_pcs = get_pitch_class_set(ref_notes)
-    res_pcs = get_pitch_class_set(res_notes)
-
-    correct_pcs = ref_pcs & res_pcs
-    missing_pcs = ref_pcs - res_pcs
-    extra_pcs   = res_pcs - ref_pcs
-
-    C = len(correct_pcs)
-    I = len(extra_pcs)
-    ref_size = len(ref_pcs)
-
+    ref_counts = Counter(note["pitch"] for note in ref_notes)
+    res_counts = Counter(note["pitch"] for note in res_notes)
+
+    correct_pitches = []
+    missing_pitches = []
+    extra_pitches = []
+
+    # For every distinct pitch on either side, match up copies one-to-one.
+    # Any leftover ref copies are missing; any leftover res copies are extra.
+    all_pitches = sorted(set(ref_counts) | set(res_counts))
+    for pitch in all_pitches:
+        matched = min(ref_counts[pitch], res_counts[pitch])
+        correct_pitches.extend([pitch] * matched)
+        if ref_counts[pitch] > matched:
+            missing_pitches.extend(
+                [pitch] * (ref_counts[pitch] - matched)
+            )
+        if res_counts[pitch] > matched:
+            extra_pitches.extend(
+                [pitch] * (res_counts[pitch] - matched)
+            )
+
+    C = len(correct_pitches)
+    I = len(extra_pitches)
+    ref_size = len(ref_notes)   # total note count, repeats included
+
     if ref_size == 0:
         accuracy = 0.0
     else:
         accuracy = (C - I + ref_size) / (2.0 * ref_size)
         accuracy = max(0.0, min(1.0, accuracy))
- 
-    return accuracy, sorted(correct_pcs), sorted(missing_pcs), sorted(extra_pcs)
+
+    return accuracy, correct_pitches, missing_pitches, extra_pitches
 
 
 # Step 0 - make first note start at t = 0.0
@@ -150,16 +168,15 @@ def normalize_start_times(notes):
     if not notes:
         return []
 
-    first_start = notes[0]["start"]
+    # Use the earliest onset rather than assuming notes[0] is the first note.
+    first_start = min(note["start"] for note in notes)
 
     shifted_notes = []
     for note in notes:
-        # Create a copy of the note dict with the "start" time shifted
-        note_copy = {
-            "pitch": note["pitch"],
-            "start": note["start"] - first_start,
-            "duration": note["duration"],
-        }
+        # Copy the complete dictionary so that additional metadata is retained.
+        note_copy = note.copy()
+        # Only modify the copied note's start time.
+        note_copy["start"] = note["start"] - first_start
         shifted_notes.append(note_copy)
 
     return shifted_notes
@@ -246,86 +263,89 @@ def group_notes_into_events(notes, chord_onset_window=DEFAULT_CHORD_ONSET_WINDOW
 
 # Step 1 -- edit-distance alignment to identify missing/extra notes and pitch errors
 # ------------------------------------------------------------------------------
-def compute_note_cost(note1, note2):
+def build_cost_matrix(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENALTY):
     """
-    Cost of aligning (replacing) one note with another, based on pitch.
-
-    cost = 0: pitches are identical (a 'match'). 
-    cost > 0: different pitches (a 'replacement')
- 
+    Precompute the full (N x M) event-alignment cost matrix using vectorised
+    NumPy broadcasting, instead of compute event cost one at a
+    time inside a Python loop. This also removes the redundant cost
+    computation that used to happen a second time during backtracking.
+
     Args:
-        note1: dict with keys "pitch" (int), "start" (float), "duration" (float)
-        note2: dict with keys "pitch" (int), "start" (float), "duration" (float)
-
+        response_events: list of event dicts (from group_notes_into_events)
+        ref_events: list of event dicts (from group_notes_into_events)
+        gap_penalty: cost used when event types (note vs chord) do not match
+
     Returns:
-        int: cost value >= 0 (lower means more similar pitch)
+        numpy array of shape (N, M): cost of aligning each response event
+        to each reference event.
     """
-    return int(abs(note1["pitch"] - note2["pitch"]))
+    N = len(response_events)
+    M = len(ref_events)
 
+    # Build simple per-event arrays: is this event a chord, and (if it is a
+    # single note) what is its pitch.
+    res_is_chord = np.array([event["event_type"] == "chord" for event in response_events])
+    ref_is_chord = np.array([event["event_type"] == "chord" for event in ref_events])
+
+    res_pitch = np.array([
+        event["notes"][0]["pitch"] if event["event_type"] == "note" else 0
+        for event in response_events
+    ])
+    ref_pitch = np.array([
+        event["notes"][0]["pitch"] if event["event_type"] == "note" else 0
+        for event in ref_events
+    ])
+
+    # Note-vs-note cost: vectorised absolute pitch difference for every pair.
+    # Shape (N, 1) - shape (1, M) broadcasts to (N, M)
+    note_cost_matrix = np.abs(res_pitch.reshape(N, 1) - ref_pitch.reshape(1, M))
+
+    # Chord-vs-chord cost: build a 12-dim binary pitch-class vector per chord,
+    # then compute Hamming distance for every pair using broadcasting.
+    res_pitch_vector = np.zeros((N, 12), dtype=int)
+    for i in range(N):
+        if res_is_chord[i]:
+            for note in response_events[i]["notes"]:
+                res_pitch_vector[i, note["pitch"] % 12] = 1
+
+    ref_pitch_vector = np.zeros((M, 12), dtype=int)
+    for j in range(M):
+        if ref_is_chord[j]:
+            for note in ref_events[j]["notes"]:
+                ref_pitch_vector[j, note["pitch"] % 12] = 1
+
+    # (N, 1, 12) vs (1, M, 12) broadcasts to (N, M, 12); summing the last
+    # axis gives the Hamming distance for every (response, ref) pair at once.
+    differs = res_pitch_vector.reshape(N, 1, 12) != ref_pitch_vector.reshape(1, M, 12)
+    chord_cost_matrix = differs.sum(axis=2)
+
+    # Type-mismatch mask: True where one side is a note and the other a chord.
+    type_mismatch = res_is_chord.reshape(N, 1) != ref_is_chord.reshape(1, M)
+    both_chords = res_is_chord.reshape(N, 1) & ref_is_chord.reshape(1, M)
+
+    # Combine the three cases -- np.where(condition, true, false)
+    cost_matrix = np.where(
+        type_mismatch,
+        gap_penalty,
+        np.where(both_chords, chord_cost_matrix, note_cost_matrix),
+    )
 
-def compute_event_cost(event1, event2, gap_penalty=DEFAULT_GAP_PENALTY):
-    """
-    Cost of aligning (substituting) one event with another.
-
-    Rules:
-      - note vs note: absolute pitch difference 
-      - chord vs chord: Hamming distance on 12-dim pitch class binary vectors
-      - note vs chord (type mismatch): return gap_penalty so the aligner
-        treats them as unaligned (insertion + deletion is preferred)
-
-    For chord vs chord, the Hamming distance counts how many of the 12
-    pitch classes differ between the two chords (symmetric difference size).
-
-    Args:
-        event1: event dict (from group_notes_into_events)
-        event2: event dict (from group_notes_into_events)
-        gap_penalty: cost of an unaligned event; used as the type-mismatch cost
-
-    Returns:
-        int: alignment cost >= 0
-    """
-    type1 = event1["event_type"]
-    type2 = event2["event_type"]
-
-    # Type mismatch: note vs chord or chord vs note.
-    # Return gap_penalty so alignment prefers to leave them unmatched.
-    if type1 != type2:
-        return gap_penalty
-
-    # Both are single notes: use pitch difference (same as Phase 1)
-    elif type1 == "note" and type2 == "note":
-        return compute_note_cost(event1["notes"][0], event2["notes"][0])
-
-    # Both are chords: Hamming distance on 12-dimensional pitch class vectors.
-    # i.e. count how many of the 12 pitch classes differ between the two chords.
-    else:
-        vec1 = [0] * 12
-        vec2 = [0] * 12
-        for note in event1["notes"]:
-            vec1[note["pitch"] % 12] = 1
-        for note in event2["notes"]:
-            vec2[note["pitch"] % 12] = 1
-        hamming = sum(1 for i in range(12) if vec1[i] != vec2[i])
-        return hamming
+    return cost_matrix.astype(int)
 
 
 def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENALTY):
     """
-    Align events (notes or chords) using edit distance (ED). 
-    The ED allows for insertions and deletions, which can be useful for 
-    evaluating musical practice containing missing/extra notes.
-    
+    Same algorithm as the production event_alignment_ED(), but the N x M
+    substitution costs are looked up from a precomputed cost matrix instead
+    of being recomputed on every DP cell and again during backtracking.
+
     Args:
         response_events: list of event dicts from group_notes_into_events
         ref_events: list of event dicts from group_notes_into_events
         gap_penalty: cost of leaving an event unaligned (insertion/deletion)
- 
+
     Returns:
-        operations: list of transformation ops dicts, in order from first event to last:
-            {'type': 'match' or 'replacement' or 'missing' or 'extra', 
-            'response_idx': int or None, 
-            'reference_idx': int or None, 
-            'cost': int}
+        operations: list of operation dicts, same format as event_alignment_ED()
         D: accumulated cost matrix, shape (N+1, M+1)
     """
     # if a raw note dict with "pitch"/"start"/"duration" but no "event_type" is
@@ -340,6 +360,9 @@ def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENA
     # the columns of D correspond to reference events
     M = len(ref_events)
 
+    # Precompute every substitution cost once
+    cost_matrix = build_cost_matrix(response_events, ref_events, gap_penalty)
+
     # Build the accumulated cost matrix D of size (N+1 x M+1)
     D = np.zeros((N + 1, M + 1), dtype=int)
 
@@ -353,7 +376,7 @@ def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENA
     # Recursion (accumulated cost / score matrix D):
     for n in range(1, N + 1):
         for m in range(1, M + 1):
-            replace_cost = compute_event_cost(response_events[n-1], ref_events[m-1], gap_penalty)
+            replace_cost = cost_matrix[n - 1, m - 1]
             D[n, m] = min(
                 D[n-1, m-1] + replace_cost, # diagonal: match or replacement
                 D[n-1, m] + gap_penalty, # vertical: extra event response[n-1]
@@ -386,11 +409,11 @@ def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENA
             n -= 1
         # For all other cases, we can move in any direction (diagonal, vertical, horizontal)
         else:
-            replace_cost = compute_event_cost(response_events[n - 1], ref_events[m - 1], gap_penalty)
-            diag = D[n - 1, m - 1] + replace_cost # diagonal: match or replacement
-            up = D[n - 1, m] + gap_penalty # vertical: extra event response[n-1]
-            left = D[n, m - 1] + gap_penalty # horizontal: missing response for ref[m-1]
-            min_cost = min(diag, up, left) # find the minimum cost step
+            replace_cost = cost_matrix[n - 1, m - 1]
+            diag = D[n - 1, m - 1] + replace_cost # Diagonal -> match or replacement
+            up = D[n - 1, m] + gap_penalty # Vertical -> response[n-1] is extra (insertion)
+            left = D[n, m - 1] + gap_penalty # Horizontal -> response is missing for ref[m-1] (deletion)
+            min_cost = min(diag, up, left) 
 
             # classify the transformation ops based on the minimum cost step
             # rule: always prefer diagonal > insertion or deletion
@@ -399,7 +422,7 @@ def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENA
                     "type": "match" if replace_cost == 0 else "replacement",
                     "response_idx": n - 1,
                     "reference_idx": m - 1,
-                    "cost": replace_cost,
+                    "cost": int(replace_cost),
                 })
                 n, m = n - 1, m - 1
             elif min_cost == up: # Vertical -> response[n-1] is extra (insertion)
@@ -419,7 +442,7 @@ def event_alignment_ED(response_events, ref_events, gap_penalty=DEFAULT_GAP_PENA
                 })
                 m -= 1
 
-    operations.reverse()  # Reverse to get ops in order from first note to last
+    operations.reverse()
     return operations, D
 
 
@@ -1006,10 +1029,10 @@ def generate_feedback_message(event_details, response_events, ref_events, stats,
                 f"{accuracy_pct}% accurate. "
             )
             if ch["missing_pitches"]:
-                missing_names = [PITCH_CLASS_NAMES[pc] for pc in ch["missing_pitches"]]
+                missing_names = [PITCH_CLASS_NAMES[pitch % 12] for pitch in ch["missing_pitches"]]
                 message = message + "Missing note(s): " + ", ".join(missing_names) + ". "
             if ch["extra_pitches"]:
-                extra_names = [PITCH_CLASS_NAMES[pc] for pc in ch["extra_pitches"]]
+                extra_names = [PITCH_CLASS_NAMES[pitch % 12] for pitch in ch["extra_pitches"]]
                 message = message + "Extra note(s) played: " + ", ".join(extra_names) + "."
             chord_detail_messages.append(message)
     # Local timing errors for chords