Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
.python-version

# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
Expand Down Expand Up @@ -194,3 +194,8 @@ score_audio_Bach-fugue_bwv_858.wav
score_audio_mozart_k265_var1.wav
performance_audio*/
score_audio*/
notebook/

.claude/
src/
scripts/
28 changes: 22 additions & 6 deletions matchmaker/matchmaker.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,4 @@
import csv
import os
from pathlib import Path
from typing import Optional, Union

import numpy as np
Expand All @@ -22,7 +20,6 @@
from matchmaker.io.audio import AudioStream
from matchmaker.io.midi import MidiStream
from matchmaker.prob.hmm import (
CosineExpGaussianAudioPitchTempoObservationModel,
GaussianAudioPitchHMM,
GaussianAudioPitchTempoHMM,
PitchIOIHMM,
Expand All @@ -37,6 +34,7 @@
from matchmaker.utils.misc import (
adjust_tempo_for_performance_audio,
generate_score_audio,
get_tempo_from_score,
is_audio_file,
is_midi_file,
save_debug_results,
Expand Down Expand Up @@ -79,6 +77,12 @@ class Matchmaker(object):
device_name_or_index : Union[str, int]
Name or index of the audio device to be used.
Ignored if `file_path` is given.
tempo : float, optional
Tempo in BPM. If None, reads from score; if score has no tempo marking,
defaults to 120 BPM.
adjust_tempo : bool (default: False)
If True and performance_file is provided, adjusts tempo based on
performance audio analysis. Applies to all methods.

"""

Expand All @@ -94,6 +98,8 @@ def __init__(
device_name_or_index: Union[str, int] = None,
sample_rate: int = SAMPLE_RATE,
frame_rate: int = FRAME_RATE,
tempo: Optional[float] = None,
adjust_tempo: bool = False,
):
self.score_file = str(score_file)
self.performance_file = (
Expand All @@ -109,9 +115,9 @@ def __init__(
self.stream = None
self.score_follower = None
self.reference_features = None
self.tempo = DEFAULT_TEMPO # bpm for quarter note
self._has_run = False
self.method = method
self.adjust_tempo = adjust_tempo

# setup score file
if score_file is None:
Expand All @@ -123,6 +129,16 @@ def __init__(
except Exception as e:
raise ValueError(f"Invalid score file: {e}")

# Set tempo: user-provided > score marking > default (120 BPM)
# _user_specified_tempo: if True, use uniform tempo; if False, use score tempo map
if tempo is not None:
self.tempo = float(tempo)
self._user_specified_tempo = True
else:
self._user_specified_tempo = False
score_tempo = get_tempo_from_score(self.score_part, self.score_file)
self.tempo = score_tempo if score_tempo is not None else DEFAULT_TEMPO

# setup feature processor
if self.feature_type is None:
self.feature_type = "chroma" if input_type == "audio" else "pitchclass"
Expand Down Expand Up @@ -238,8 +254,8 @@ def __init__(

def preprocess_score(self):
if self.input_type == "audio":
if self.performance_file is not None:
# tempo is slightly adjusted to reflect the tempo of the performance audio
# Adjust tempo based on performance audio if requested
if self.adjust_tempo and self.performance_file is not None:
self.tempo = adjust_tempo_for_performance_audio(
self.score_part, self.performance_file, self.tempo
)
Expand Down
239 changes: 235 additions & 4 deletions matchmaker/utils/misc.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,24 +7,147 @@
import csv
import numbers
import os
import re
import xml.etree.ElementTree as ET
from pathlib import Path
from queue import Empty, Queue
from typing import Any, Dict, Iterable, List, Union
from typing import Any, Dict, Iterable, List, Optional, Union

import librosa
import mido
import numpy as np
import partitura
import scipy
import soundfile as sf
from matplotlib import pyplot as plt
from numpy.typing import NDArray
from partitura.io.exportmidi import get_ppq
from partitura.score import ScoreLike
from partitura.utils.music import performance_notearray_from_score_notearray

from matchmaker.features.audio import SAMPLE_RATE

# Tempo marking to BPM mapping
# Reference: https://en.wikipedia.org/wiki/Tempo#Basic_tempo_markings
TEMPO_MARKING_TO_BPM = {
# Very slow (24-40 BPM)
"larghissimo": 24,
"grave": 30,
# Slow (40-66 BPM)
"largo": 40,
"larghetto": 50,
"lento": 60,
# Slow-moderate (44-80 BPM)
"adagio": 60,
"adagietto": 70,
# Walking pace (56-108 BPM)
"andante": 80,
"andantino": 90,
# Moderate (86-126 BPM)
"moderato": 110,
"allegretto": 120,
# Fast (100-156 BPM)
"allegro": 130,
# Very fast (136-200+ BPM)
"vivace": 150,
"vivacissimo": 170,
"presto": 180,
"prestissimo": 200,
}


def extract_tempo_marking_from_musicxml(
score_file: Union[str, Path],
) -> Optional[float]:
"""
Extract tempo from text tempo marking (e.g., "Allegro", "Andante") in MusicXML.

Parses <direction-type><words>...</words></direction-type> elements and
matches against known tempo markings. The tempo is adjusted based on the
time signature to return quarter-note BPM.

Handles both simple and compound meters:
- Simple meters (2/2, 4/4, etc.): beat is the note indicated by denominator
- Compound meters (6/8, 9/8, 12/8): beat is a dotted note (3 subdivisions)

For example:
- 2/2 "Allegro" (130 half notes/min) → 260 quarter-note BPM
- 6/8 "Presto" (180 dotted-quarters/min) → 270 quarter-note BPM

Parameters
----------
score_file : str or Path
Path to the MusicXML file

Returns
-------
float or None
Quarter-note BPM based on tempo marking, or None if not found
"""
try:
tree = ET.parse(str(score_file))
root = tree.getroot()

# Track current time signature
# Default to 4/4 (simple meter, quarter note beat)
current_beats = 4
current_beat_type = 4

# Process measures in order to track time signature changes
for part in root.iter("part"):
for measure in part.iter("measure"):
# Check for time signature change in this measure
for attributes in measure.iter("attributes"):
for time_elem in attributes.iter("time"):
beats_elem = time_elem.find("beats")
beat_type_elem = time_elem.find("beat-type")
if beats_elem is not None and beats_elem.text:
current_beats = int(beats_elem.text)
if beat_type_elem is not None and beat_type_elem.text:
current_beat_type = int(beat_type_elem.text)

# Look for tempo marking in this measure
for direction in measure.iter("direction"):
for direction_type in direction.iter("direction-type"):
for words in direction_type.iter("words"):
if words.text:
text = words.text.lower().strip()
# Check each tempo marking
for marking, bpm in TEMPO_MARKING_TO_BPM.items():
# Match if marking appears at the start of the text
if re.match(rf"^{marking}\b", text):
# Check if compound meter (6/8, 9/8, 12/8, etc.)
# Compound meter: numerator divisible by 3 and >= 6
is_compound = (
current_beats >= 6
and current_beats % 3 == 0
)

if is_compound:
# Compound meter: beat is dotted note
# e.g., 6/8: dotted quarter = 3 eighth notes
# To maintain the same "feel" as simple meter:
# - In 4/4 "Andante=80": quarter = 0.75s
# - In 6/8 we want dotted quarter as beat
# - dotted quarter = 1.5 × quarter
# - So dotted quarter BPM = quarter BPM / 1.5
quarter_note_bpm = bpm / 1.5
else:
# Simple meter: beat-type indicates beat note
# Convert to quarter-note BPM
# beat-type 2 (half note): multiply by 2
# beat-type 4 (quarter note): no change
# beat-type 8 (eighth note): divide by 2
quarter_note_bpm = bpm * (
4.0 / current_beat_type
)

return float(quarter_note_bpm)

# Only process first part to avoid duplicates
break
except Exception:
pass

return None


class MatchmakerInvalidParameterTypeError(Exception):
"""
Expand Down Expand Up @@ -191,6 +314,114 @@ def interleave_with_constant(
return interleaved_array


def get_tempo_from_score(
score_part: ScoreLike,
score_file: Optional[Union[str, Path]] = None,
) -> Optional[float]:
"""
Extract first tempo marking from score if available.

Tries multiple sources in order:
1. Partitura Tempo objects (explicit BPM)
2. MusicXML <sound tempo="..."/> element (if score_file provided)

Parameters
----------
score_part : ScoreLike
Partitura score part
score_file : str or Path, optional
Path to the score file. Used as fallback to parse MusicXML directly
when partitura doesn't extract tempo.

Returns
-------
float or None
Tempo in BPM if found in score, None otherwise.
"""
# Try partitura Tempo objects first
if score_part is not None:
try:
for tempo_obj in score_part.iter_all(partitura.score.Tempo):
if hasattr(tempo_obj, "bpm") and tempo_obj.bpm is not None:
return float(tempo_obj.bpm)
except Exception:
pass

# Fallback: parse MusicXML directly for <sound tempo="..."/>
if score_file is not None:
try:
import xml.etree.ElementTree as ET

tree = ET.parse(str(score_file))
root = tree.getroot()

for sound_elem in root.iter("sound"):
tempo_attr = sound_elem.get("tempo")
if tempo_attr is not None:
return float(tempo_attr)
except Exception:
pass

return None


def get_tempo_at_beat(
score_part: ScoreLike,
beat: float,
default_tempo: float = 120.0,
) -> float:
"""
Get tempo (BPM) at a specific beat position in the score.

Uses score tempo markings if available. Falls back to default_tempo otherwise.

Parameters
----------
score_part : ScoreLike
Partitura score part
beat : float
Beat position in the score
default_tempo : float
Default tempo to use if no tempo markings found

Returns
-------
float
Tempo in BPM at the given beat position
"""
if score_part is None:
return default_tempo

# Collect all tempo markings with their positions
tempo_changes = []
try:
for tempo_obj in score_part.iter_all(partitura.score.Tempo):
if hasattr(tempo_obj, "bpm") and tempo_obj.bpm is not None:
# Get beat position of tempo marking
start_time = getattr(tempo_obj, "start", None)
if start_time is not None:
tempo_beat = score_part.beat_map(start_time.t)
tempo_changes.append((tempo_beat, float(tempo_obj.bpm)))
except Exception:
pass

if not tempo_changes:
return default_tempo

# Sort by beat position
tempo_changes.sort(key=lambda x: x[0])

# Find the tempo at the given beat (last tempo marking before or at beat)
current_tempo = default_tempo
for tempo_beat, bpm in tempo_changes:
if tempo_beat <= beat:
current_tempo = bpm
else:
break

return current_tempo


def adjust_tempo_for_performance_audio(
score: ScoreLike, performance_audio: Path, default_tempo: int = 120
):
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ dependencies = [
"numpy>=1.26.3,<2.0",
"scipy>=1.11.4,<1.15",
"librosa>=0.10.1",
"pandas>=2.0.0",
"partitura>=1.7.0",
"progressbar2>=4.2.0",
"python-hiddenmarkov>=0.1.3",
Expand Down
10 changes: 10 additions & 0 deletions tests/resources/Bach-fugue_bwv_858.musicxml
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,16 @@
<line>4</line>
</clef>
</attributes>
<direction placement="above" system="only-top">
<direction-type>
<metronome parentheses="no" default-x="-38.52" relative-y="20">
<beat-unit>quarter</beat-unit>
<per-minute>70</per-minute>
</metronome>
</direction-type>
<staff>1</staff>
<sound tempo="70"/>
</direction>
<note id="r1">
<rest>
<display-step>A</display-step>
Expand Down