669 lines
22 KiB
Python
669 lines
22 KiB
Python
"""
|
|||
|
|
SoulX-Singer MIDI <-> metadata converter.
|
||
|
|
|
||
|
|
Converts between SoulX-Singer-style metadata JSON (with note_text, note_dur,
|
||
|
|
note_pitch, note_type per segment) and standard MIDI files. Uses an internal
|
||
|
|
Note dataclass (start_s, note_dur, note_text, note_pitch, note_type) as the
|
||
|
|
intermediate representation.
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import shutil
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Any, List, Tuple, Union
|
||
|
|
|
||
|
|
import librosa
|
||
|
|
import mido
|
||
|
|
from soundfile import write
|
||
|
|
|
||
|
|
from .f0_extraction import F0Extractor
|
||
|
|
from .g2p import g2p_transform
|
||
|
|
|
||
|
|
|
||
|
|
# Audio and segmenting constants (used by _edit_data_to_meta)
|
||
|
|
SAMPLE_RATE = 44100
|
||
|
|
DEFAULT_LANGUAGE = "Mandarin"
|
||
|
|
MAX_GAP_SEC = 5.0 # gap (sec) above which we start a new segment
|
||
|
|
MAX_SEGMENT_DUR_SUM_SEC = 60.0 # max cumulative note duration per segment (sec)
|
||
|
|
MIN_GAP_THRESHOLD_SEC = 0.001 # ignore gaps smaller than this
|
||
|
|
LONG_SILENCE_THRESHOLD_SEC = 0.05 # treat as separate <SP> if gap larger
|
||
|
|
MAX_LEADING_SP_DUR_SEC = 2.0 # cap leading silence in a segment to this (sec)
|
||
|
|
DEFAULT_RMVPE_MODEL_PATH = "pretrained_models/SoulX-Singer-Preprocess/rmvpe/rmvpe.pt"
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Note:
|
||
|
|
"""Single note: text, duration (seconds), pitch (MIDI), type. start_s is absolute start time in seconds (for ordering / MIDI)."""
|
||
|
|
start_s: float
|
||
|
|
note_dur: float
|
||
|
|
note_text: str
|
||
|
|
note_pitch: int
|
||
|
|
note_type: int
|
||
|
|
|
||
|
|
@property
|
||
|
|
def end_s(self) -> float:
|
||
|
|
return self.start_s + self.note_dur
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def remove_duplicate_segments(meta_data: List[dict]) -> None:
|
||
|
|
"""Merge consecutive identical notes (same text, pitch, type) within each segment. Mutates meta_data in place."""
|
||
|
|
for idx, segment in enumerate(meta_data):
|
||
|
|
texts = segment["note_text"]
|
||
|
|
durs = segment["note_dur"]
|
||
|
|
pitches = segment["note_pitch"]
|
||
|
|
types = segment["note_type"]
|
||
|
|
new_texts = []
|
||
|
|
new_durs = []
|
||
|
|
new_pitches = []
|
||
|
|
new_types = []
|
||
|
|
for i in range(len(texts)):
|
||
|
|
if i == 0:
|
||
|
|
new_texts.append(texts[i])
|
||
|
|
new_durs.append(durs[i])
|
||
|
|
new_pitches.append(pitches[i])
|
||
|
|
new_types.append(types[i])
|
||
|
|
continue
|
||
|
|
t, d, p, ty = texts[i], durs[i], pitches[i], types[i]
|
||
|
|
if t == "<SP>" and texts[i - 1] == "<SP>":
|
||
|
|
new_durs[-1] += d
|
||
|
|
continue
|
||
|
|
if t == texts[i - 1] and p == pitches[i - 1] and ty == types[i - 1]:
|
||
|
|
new_durs[-1] += d
|
||
|
|
else:
|
||
|
|
new_texts.append(t)
|
||
|
|
new_durs.append(d)
|
||
|
|
new_pitches.append(p)
|
||
|
|
new_types.append(ty)
|
||
|
|
meta_data[idx]["note_text"] = new_texts
|
||
|
|
meta_data[idx]["note_dur"] = new_durs
|
||
|
|
meta_data[idx]["note_pitch"] = new_pitches
|
||
|
|
meta_data[idx]["note_type"] = new_types
|
||
|
|
|
||
|
|
def meta2notes(meta_path: str) -> List[Note]:
|
||
|
|
"""Parse SoulX-Singer metadata JSON into a flat list of Note (absolute start_s)."""
|
||
|
|
with open(meta_path, "r", encoding="utf-8") as f:
|
||
|
|
segments = json.load(f)
|
||
|
|
if not isinstance(segments, list):
|
||
|
|
raise ValueError(f"Metadata must be a list of segments, got {type(segments).__name__}")
|
||
|
|
if not segments:
|
||
|
|
raise ValueError("Metadata has no segments.")
|
||
|
|
|
||
|
|
notes: List[Note] = []
|
||
|
|
for seg in segments:
|
||
|
|
offset_s = seg["time"][0] / 1000
|
||
|
|
words = [str(x).replace("<AP>", "<SP>") for i, x in enumerate(seg["text"].split())]
|
||
|
|
word_durs = [float(x) for x in seg["duration"].split()]
|
||
|
|
pitches = [int(x) for x in seg["note_pitch"].split()]
|
||
|
|
types = [int(x) if words[i] != "<SP>" else 1 for i, x in enumerate(seg["note_type"].split())]
|
||
|
|
if len(words) != len(word_durs) or len(word_durs) != len(pitches) or len(pitches) != len(types):
|
||
|
|
raise ValueError(
|
||
|
|
f"Length mismatch in segment {seg.get('item_name', '?')}: "
|
||
|
|
"note_text, note_dur, note_pitch, note_type must have same length"
|
||
|
|
)
|
||
|
|
current_s = offset_s
|
||
|
|
for text, dur, pitch, type_ in zip(words, word_durs, pitches, types):
|
||
|
|
notes.append(
|
||
|
|
Note(
|
||
|
|
start_s=current_s,
|
||
|
|
note_dur=float(dur),
|
||
|
|
note_text=str(text),
|
||
|
|
note_pitch=int(pitch),
|
||
|
|
note_type=int(type_),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
current_s += float(dur)
|
||
|
|
return notes
|
||
|
|
|
||
|
|
def _append_segment_to_meta(
|
||
|
|
meta_path_str: str,
|
||
|
|
cut_wavs_output_dir: str,
|
||
|
|
vocal_file: str,
|
||
|
|
audio_data: Any,
|
||
|
|
meta_data: List[dict],
|
||
|
|
note_start: List[float],
|
||
|
|
note_end: List[float],
|
||
|
|
note_text: List[Any],
|
||
|
|
note_pitch: List[Any],
|
||
|
|
note_type: List[Any],
|
||
|
|
note_dur: List[float],
|
||
|
|
end_time_ms_override: float | None = None,
|
||
|
|
) -> None:
|
||
|
|
"""Write one segment wav and append one segment dict to meta_data. Caller clears note_* lists after."""
|
||
|
|
base_name = os.path.splitext(os.path.basename(meta_path_str))[0]
|
||
|
|
item_name = f"{base_name}_{len(meta_data)}"
|
||
|
|
wav_fn = os.path.join(cut_wavs_output_dir, f"{item_name}.wav")
|
||
|
|
start_ms = int(note_start[0] * 1000)
|
||
|
|
end_ms = (
|
||
|
|
int(end_time_ms_override)
|
||
|
|
if end_time_ms_override is not None
|
||
|
|
else int(note_end[-1] * 1000)
|
||
|
|
)
|
||
|
|
start_sample = int(note_start[0] * SAMPLE_RATE)
|
||
|
|
end_sample = int(note_end[-1] * SAMPLE_RATE)
|
||
|
|
write(wav_fn, audio_data[start_sample:end_sample], SAMPLE_RATE)
|
||
|
|
meta_data.append({
|
||
|
|
"item_name": item_name,
|
||
|
|
"wav_fn": wav_fn,
|
||
|
|
"origin_wav_fn": vocal_file,
|
||
|
|
"start_time_ms": start_ms,
|
||
|
|
"end_time_ms": end_ms,
|
||
|
|
"language": DEFAULT_LANGUAGE,
|
||
|
|
"note_text": list(note_text),
|
||
|
|
"note_pitch": list(note_pitch),
|
||
|
|
"note_type": list(note_type),
|
||
|
|
"note_dur": list(note_dur),
|
||
|
|
})
|
||
|
|
|
||
|
|
|
||
|
|
def convert_meta(meta_data: List[dict], rmvpe_model_path, device="cuda"):
|
||
|
|
pitch_extractor = F0Extractor(rmvpe_model_path, device=device, verbose=False)
|
||
|
|
converted_data = []
|
||
|
|
|
||
|
|
for item in meta_data:
|
||
|
|
wav_fn = item.get("wav_fn")
|
||
|
|
if not wav_fn or not os.path.isfile(wav_fn):
|
||
|
|
raise FileNotFoundError(f"Segment wav file not found: {wav_fn}")
|
||
|
|
f0 = pitch_extractor.process(wav_fn)
|
||
|
|
converted_item = {
|
||
|
|
"index": item.get("item_name"),
|
||
|
|
"language": item.get("language"),
|
||
|
|
"time": [item.get("start_time_ms", 0), item.get("end_time_ms", sum(item["note_dur"]) * 1000)],
|
||
|
|
"duration": " ".join(str(round(x, 2)) for x in item.get("note_dur", [])),
|
||
|
|
"text": " ".join(item.get("note_text", [])),
|
||
|
|
"phoneme": " ".join(g2p_transform(item.get("note_text", []), DEFAULT_LANGUAGE)),
|
||
|
|
"note_pitch": " ".join(str(x) for x in item.get("note_pitch", [])),
|
||
|
|
"note_type": " ".join(str(x) for x in item.get("note_type", [])),
|
||
|
|
"f0": " ".join(str(round(float(x), 1)) for x in f0),
|
||
|
|
}
|
||
|
|
converted_data.append(converted_item)
|
||
|
|
|
||
|
|
return converted_data
|
||
|
|
|
||
|
|
|
||
|
|
def _edit_data_to_meta(
|
||
|
|
meta_path_str: str,
|
||
|
|
edit_data: List[dict],
|
||
|
|
vocal_file: str,
|
||
|
|
rmvpe_model_path: str | None = None,
|
||
|
|
device: str = "cuda",
|
||
|
|
) -> None:
|
||
|
|
"""Write SoulX-Singer metadata JSON from edit_data (list of {start, end, note_text, note_pitch, note_type})."""
|
||
|
|
# Use a fixed temporary directory for cut wavs
|
||
|
|
cut_wavs_output_dir = os.path.join(os.path.dirname(vocal_file), "cut_wavs_tmp")
|
||
|
|
os.makedirs(cut_wavs_output_dir, exist_ok=True)
|
||
|
|
|
||
|
|
note_text: List[Any] = []
|
||
|
|
note_pitch: List[Any] = []
|
||
|
|
note_type: List[Any] = []
|
||
|
|
note_dur: List[float] = []
|
||
|
|
note_start: List[float] = []
|
||
|
|
note_end: List[float] = []
|
||
|
|
prev_end = 0.0
|
||
|
|
meta_data: List[dict] = []
|
||
|
|
audio_data, _ = librosa.load(vocal_file, sr=SAMPLE_RATE, mono=True)
|
||
|
|
dur_sum = 0.0
|
||
|
|
|
||
|
|
for entry in edit_data:
|
||
|
|
start = float(entry["start"])
|
||
|
|
end = float(entry["end"])
|
||
|
|
text = entry["note_text"]
|
||
|
|
pitch = entry["note_pitch"]
|
||
|
|
type_ = entry["note_type"]
|
||
|
|
|
||
|
|
if text == "" or pitch == "" or type_ == "":
|
||
|
|
note_text.append("<SP>")
|
||
|
|
note_pitch.append(0)
|
||
|
|
note_type.append(1)
|
||
|
|
note_dur.append(end - start)
|
||
|
|
note_start.append(start)
|
||
|
|
note_end.append(end)
|
||
|
|
prev_end = end
|
||
|
|
dur_sum += end - start
|
||
|
|
continue
|
||
|
|
|
||
|
|
if (
|
||
|
|
len(note_text) > 0
|
||
|
|
and note_text[-1] == "<SP>"
|
||
|
|
and note_dur[-1] > MAX_LEADING_SP_DUR_SEC
|
||
|
|
):
|
||
|
|
cut_time = note_dur[-1] - MAX_LEADING_SP_DUR_SEC
|
||
|
|
note_dur[-1] = MAX_LEADING_SP_DUR_SEC
|
||
|
|
end_ms_override = note_end[-1] * 1000 - cut_time * 1000
|
||
|
|
_append_segment_to_meta(
|
||
|
|
meta_path_str,
|
||
|
|
cut_wavs_output_dir,
|
||
|
|
vocal_file,
|
||
|
|
audio_data,
|
||
|
|
meta_data,
|
||
|
|
note_start,
|
||
|
|
note_end,
|
||
|
|
note_text,
|
||
|
|
note_pitch,
|
||
|
|
note_type,
|
||
|
|
note_dur,
|
||
|
|
end_time_ms_override=end_ms_override,
|
||
|
|
)
|
||
|
|
note_text = []
|
||
|
|
note_pitch = []
|
||
|
|
note_type = []
|
||
|
|
note_dur = []
|
||
|
|
note_start = []
|
||
|
|
note_end = []
|
||
|
|
prev_end = start
|
||
|
|
dur_sum = 0.0
|
||
|
|
|
||
|
|
gap_from_prev = start - prev_end
|
||
|
|
gap_from_last_note = (start - note_end[-1]) if note_end else 0.0
|
||
|
|
if (
|
||
|
|
gap_from_prev >= MAX_GAP_SEC
|
||
|
|
or gap_from_last_note >= MAX_GAP_SEC
|
||
|
|
or dur_sum >= MAX_SEGMENT_DUR_SUM_SEC
|
||
|
|
):
|
||
|
|
if len(note_text) > 0:
|
||
|
|
_append_segment_to_meta(
|
||
|
|
meta_path_str,
|
||
|
|
cut_wavs_output_dir,
|
||
|
|
vocal_file,
|
||
|
|
audio_data,
|
||
|
|
meta_data,
|
||
|
|
note_start,
|
||
|
|
note_end,
|
||
|
|
note_text,
|
||
|
|
note_pitch,
|
||
|
|
note_type,
|
||
|
|
note_dur,
|
||
|
|
)
|
||
|
|
note_text = []
|
||
|
|
note_pitch = []
|
||
|
|
note_type = []
|
||
|
|
note_dur = []
|
||
|
|
note_start = []
|
||
|
|
note_end = []
|
||
|
|
prev_end = start
|
||
|
|
dur_sum = 0.0
|
||
|
|
|
||
|
|
if start - prev_end > MIN_GAP_THRESHOLD_SEC:
|
||
|
|
if start - prev_end > LONG_SILENCE_THRESHOLD_SEC or len(note_text) == 0:
|
||
|
|
note_text.append("<SP>")
|
||
|
|
note_pitch.append(0)
|
||
|
|
note_type.append(1)
|
||
|
|
note_dur.append(start - prev_end)
|
||
|
|
note_start.append(prev_end)
|
||
|
|
note_end.append(start)
|
||
|
|
else:
|
||
|
|
if len(note_dur) > 0:
|
||
|
|
note_dur[-1] += start - prev_end
|
||
|
|
note_end[-1] = start
|
||
|
|
|
||
|
|
prev_end = end
|
||
|
|
note_text.append(text)
|
||
|
|
note_pitch.append(int(pitch))
|
||
|
|
note_type.append(int(type_))
|
||
|
|
note_dur.append(end - start)
|
||
|
|
note_start.append(start)
|
||
|
|
note_end.append(end)
|
||
|
|
dur_sum += end - start
|
||
|
|
|
||
|
|
if len(note_text) > 0:
|
||
|
|
_append_segment_to_meta(
|
||
|
|
meta_path_str,
|
||
|
|
cut_wavs_output_dir,
|
||
|
|
vocal_file,
|
||
|
|
audio_data,
|
||
|
|
meta_data,
|
||
|
|
note_start,
|
||
|
|
note_end,
|
||
|
|
note_text,
|
||
|
|
note_pitch,
|
||
|
|
note_type,
|
||
|
|
note_dur,
|
||
|
|
)
|
||
|
|
|
||
|
|
remove_duplicate_segments(meta_data)
|
||
|
|
|
||
|
|
_rmvpe_path = rmvpe_model_path or DEFAULT_RMVPE_MODEL_PATH
|
||
|
|
converted_data = convert_meta(meta_data, _rmvpe_path, device)
|
||
|
|
|
||
|
|
with open(meta_path_str, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(converted_data, f, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
# Clean up temporary cut wavs directory
|
||
|
|
try:
|
||
|
|
shutil.rmtree(cut_wavs_output_dir, ignore_errors=True)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def notes2meta(
|
||
|
|
notes: List[Note],
|
||
|
|
meta_path: str,
|
||
|
|
vocal_file: str,
|
||
|
|
rmvpe_model_path: str | None = None,
|
||
|
|
device: str = "cuda",
|
||
|
|
) -> None:
|
||
|
|
"""Write SoulX-Singer metadata JSON from a list of Note (segmenting + wav cuts)."""
|
||
|
|
edit_data = [
|
||
|
|
{
|
||
|
|
"start": n.start_s,
|
||
|
|
"end": n.end_s,
|
||
|
|
"note_text": n.note_text,
|
||
|
|
"note_pitch": str(n.note_pitch),
|
||
|
|
"note_type": str(n.note_type),
|
||
|
|
}
|
||
|
|
for n in notes
|
||
|
|
]
|
||
|
|
_edit_data_to_meta(
|
||
|
|
str(meta_path),
|
||
|
|
edit_data,
|
||
|
|
vocal_file,
|
||
|
|
rmvpe_model_path=rmvpe_model_path,
|
||
|
|
device=device,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class MidiDefaults:
|
||
|
|
ticks_per_beat: int = 500
|
||
|
|
tempo: int = 500000 # microseconds per beat (120 BPM)
|
||
|
|
time_signature: Tuple[int, int] = (4, 4)
|
||
|
|
velocity: int = 64
|
||
|
|
|
||
|
|
|
||
|
|
def _seconds_to_ticks(seconds: float, ticks_per_beat: int, tempo: int) -> int:
|
||
|
|
return int(round(seconds * ticks_per_beat * 1_000_000 / tempo))
|
||
|
|
|
||
|
|
|
||
|
|
def notes2midi(
|
||
|
|
notes: List[Note],
|
||
|
|
midi_path: str,
|
||
|
|
defaults: MidiDefaults | None = None,
|
||
|
|
) -> None:
|
||
|
|
"""Write MIDI file from a list of Note."""
|
||
|
|
defaults = defaults or MidiDefaults()
|
||
|
|
if not notes:
|
||
|
|
raise ValueError("Empty note list.")
|
||
|
|
|
||
|
|
events: List[Tuple[int, int, Union[mido.Message, mido.MetaMessage]]] = []
|
||
|
|
for n in notes:
|
||
|
|
start_s = n.start_s
|
||
|
|
end_s = n.end_s
|
||
|
|
if end_s <= start_s:
|
||
|
|
continue
|
||
|
|
|
||
|
|
start_ticks = _seconds_to_ticks(
|
||
|
|
start_s, defaults.ticks_per_beat, defaults.tempo
|
||
|
|
)
|
||
|
|
end_ticks = _seconds_to_ticks(
|
||
|
|
end_s, defaults.ticks_per_beat, defaults.tempo
|
||
|
|
)
|
||
|
|
if end_ticks <= start_ticks:
|
||
|
|
end_ticks = start_ticks + 1
|
||
|
|
|
||
|
|
lyric = n.note_text
|
||
|
|
try:
|
||
|
|
lyric = lyric.encode("utf-8").decode("latin1")
|
||
|
|
except (UnicodeEncodeError, UnicodeDecodeError):
|
||
|
|
pass
|
||
|
|
if n.note_type == 3:
|
||
|
|
lyric = "-"
|
||
|
|
|
||
|
|
events.append(
|
||
|
|
(start_ticks, 1, mido.MetaMessage("lyrics", text=lyric, time=0))
|
||
|
|
)
|
||
|
|
events.append(
|
||
|
|
(
|
||
|
|
start_ticks,
|
||
|
|
2,
|
||
|
|
mido.Message(
|
||
|
|
"note_on",
|
||
|
|
note=n.note_pitch,
|
||
|
|
velocity=defaults.velocity,
|
||
|
|
time=0,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
events.append(
|
||
|
|
(
|
||
|
|
end_ticks,
|
||
|
|
0,
|
||
|
|
mido.Message("note_off", note=n.note_pitch, velocity=0, time=0),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
events.sort(key=lambda x: (x[0], x[1]))
|
||
|
|
|
||
|
|
mid = mido.MidiFile(ticks_per_beat=defaults.ticks_per_beat)
|
||
|
|
track = mido.MidiTrack()
|
||
|
|
mid.tracks.append(track)
|
||
|
|
|
||
|
|
track.append(mido.MetaMessage("set_tempo", tempo=defaults.tempo, time=0))
|
||
|
|
track.append(
|
||
|
|
mido.MetaMessage(
|
||
|
|
"time_signature",
|
||
|
|
numerator=defaults.time_signature[0],
|
||
|
|
denominator=defaults.time_signature[1],
|
||
|
|
time=0,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
last_tick = 0
|
||
|
|
for tick, _, msg in events:
|
||
|
|
msg.time = max(0, tick - last_tick)
|
||
|
|
track.append(msg)
|
||
|
|
last_tick = tick
|
||
|
|
|
||
|
|
track.append(mido.MetaMessage("end_of_track", time=0))
|
||
|
|
mid.save(midi_path)
|
||
|
|
|
||
|
|
|
||
|
|
def midi2notes(midi_path: str) -> List[Note]:
|
||
|
|
"""Parse MIDI file into a list of Note. Merges all tracks; tempo from last set_tempo event."""
|
||
|
|
mid = mido.MidiFile(midi_path)
|
||
|
|
ticks_per_beat = mid.ticks_per_beat
|
||
|
|
tempo = 500000
|
||
|
|
|
||
|
|
raw_notes: List[dict] = []
|
||
|
|
lyrics: List[Tuple[int, str]] = []
|
||
|
|
|
||
|
|
for track in mid.tracks:
|
||
|
|
abs_ticks = 0
|
||
|
|
active = {}
|
||
|
|
for msg in track:
|
||
|
|
abs_ticks += msg.time
|
||
|
|
if msg.type == "set_tempo":
|
||
|
|
tempo = msg.tempo
|
||
|
|
elif msg.type == "lyrics":
|
||
|
|
text = msg.text
|
||
|
|
try:
|
||
|
|
text = text.encode("latin1").decode("utf-8")
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
lyrics.append((abs_ticks, text))
|
||
|
|
elif msg.type == "note_on":
|
||
|
|
key = (msg.channel, msg.note)
|
||
|
|
if msg.velocity > 0:
|
||
|
|
active[key] = (abs_ticks, msg.velocity)
|
||
|
|
else:
|
||
|
|
if key in active:
|
||
|
|
start_ticks, vel = active.pop(key)
|
||
|
|
raw_notes.append(
|
||
|
|
{
|
||
|
|
"midi": msg.note,
|
||
|
|
"start_ticks": start_ticks,
|
||
|
|
"duration_ticks": abs_ticks - start_ticks,
|
||
|
|
"velocity": vel,
|
||
|
|
"lyric": "",
|
||
|
|
}
|
||
|
|
)
|
||
|
|
elif msg.type == "note_off":
|
||
|
|
key = (msg.channel, msg.note)
|
||
|
|
if key in active:
|
||
|
|
start_ticks, vel = active.pop(key)
|
||
|
|
raw_notes.append(
|
||
|
|
{
|
||
|
|
"midi": msg.note,
|
||
|
|
"start_ticks": start_ticks,
|
||
|
|
"duration_ticks": abs_ticks - start_ticks,
|
||
|
|
"velocity": vel,
|
||
|
|
"lyric": "",
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
if not raw_notes:
|
||
|
|
raise ValueError("No notes found in MIDI file")
|
||
|
|
|
||
|
|
for n in raw_notes:
|
||
|
|
n["end_ticks"] = n["start_ticks"] + n["duration_ticks"]
|
||
|
|
|
||
|
|
raw_notes.sort(key=lambda n: n["start_ticks"])
|
||
|
|
lyrics.sort(key=lambda x: x[0])
|
||
|
|
|
||
|
|
trimmed = []
|
||
|
|
for note in raw_notes:
|
||
|
|
while trimmed:
|
||
|
|
prev = trimmed[-1]
|
||
|
|
if note["start_ticks"] < prev["end_ticks"]:
|
||
|
|
prev["end_ticks"] = note["start_ticks"]
|
||
|
|
prev["duration_ticks"] = prev["end_ticks"] - prev["start_ticks"]
|
||
|
|
if prev["duration_ticks"] <= 0:
|
||
|
|
trimmed.pop()
|
||
|
|
continue
|
||
|
|
break
|
||
|
|
trimmed.append(note)
|
||
|
|
raw_notes = trimmed
|
||
|
|
|
||
|
|
tolerance = ticks_per_beat // 100
|
||
|
|
lyric_idx = 0
|
||
|
|
for note in raw_notes:
|
||
|
|
while lyric_idx < len(lyrics) and lyrics[lyric_idx][0] < note["start_ticks"] - tolerance:
|
||
|
|
lyric_idx += 1
|
||
|
|
if lyric_idx < len(lyrics):
|
||
|
|
lyric_ticks, lyric_text = lyrics[lyric_idx]
|
||
|
|
if abs(lyric_ticks - note["start_ticks"]) <= tolerance:
|
||
|
|
note["lyric"] = lyric_text
|
||
|
|
lyric_idx += 1
|
||
|
|
|
||
|
|
def ticks_to_seconds(ticks: int) -> float:
|
||
|
|
return (ticks / ticks_per_beat) * (tempo / 1_000_000)
|
||
|
|
|
||
|
|
result: List[Note] = []
|
||
|
|
prev_end_s = 0.0
|
||
|
|
for idx, n in enumerate(raw_notes):
|
||
|
|
start_s = ticks_to_seconds(n["start_ticks"])
|
||
|
|
end_s = ticks_to_seconds(n["end_ticks"])
|
||
|
|
if prev_end_s > start_s:
|
||
|
|
start_s = prev_end_s
|
||
|
|
dur_s = end_s - start_s
|
||
|
|
if dur_s <= 0:
|
||
|
|
continue
|
||
|
|
|
||
|
|
lyric = n.get("lyric", "")
|
||
|
|
if not lyric:
|
||
|
|
tp = 2
|
||
|
|
text = "啦"
|
||
|
|
elif lyric == "<SP>":
|
||
|
|
tp = 1
|
||
|
|
text = "<SP>"
|
||
|
|
elif lyric == "-":
|
||
|
|
tp = 3
|
||
|
|
text = raw_notes[idx - 1].get("lyric", "-") if idx > 0 else "-"
|
||
|
|
else:
|
||
|
|
tp = 2
|
||
|
|
text = lyric
|
||
|
|
|
||
|
|
result.append(
|
||
|
|
Note(
|
||
|
|
start_s=start_s,
|
||
|
|
note_dur=dur_s,
|
||
|
|
note_text=text,
|
||
|
|
note_pitch=n["midi"],
|
||
|
|
note_type=tp,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
prev_end_s = end_s
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def meta2midi(meta_path: str, midi_path: str, defaults: MidiDefaults | None = None) -> None:
|
||
|
|
"""Convert SoulX-Singer metadata JSON to MIDI file (meta -> List[Note] -> midi)."""
|
||
|
|
notes = meta2notes(meta_path)
|
||
|
|
notes2midi(notes, midi_path, defaults)
|
||
|
|
print(f"Saved MIDI to {midi_path}")
|
||
|
|
|
||
|
|
|
||
|
|
def midi2meta(
|
||
|
|
midi_path: str,
|
||
|
|
meta_path: str,
|
||
|
|
vocal_file: str,
|
||
|
|
rmvpe_model_path: str | None = None,
|
||
|
|
device: str = "cuda",
|
||
|
|
) -> None:
|
||
|
|
"""Convert MIDI file to SoulX-Singer metadata JSON (midi -> List[Note] -> meta)."""
|
||
|
|
meta_dir = os.path.dirname(meta_path)
|
||
|
|
if meta_dir:
|
||
|
|
os.makedirs(meta_dir, exist_ok=True)
|
||
|
|
# cut_wavs will be written to a fixed temporary directory inside _edit_data_to_meta
|
||
|
|
notes = midi2notes(midi_path)
|
||
|
|
notes2meta(
|
||
|
|
notes,
|
||
|
|
meta_path,
|
||
|
|
vocal_file,
|
||
|
|
rmvpe_model_path=rmvpe_model_path,
|
||
|
|
device=device,
|
||
|
|
)
|
||
|
|
print(f"Saved Meta to {meta_path}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
import argparse
|
||
|
|
|
||
|
|
parser = argparse.ArgumentParser(
|
||
|
|
description="Convert SoulX-Singer metadata JSON <-> MIDI."
|
||
|
|
)
|
||
|
|
parser.add_argument("--meta", type=str, help="Path to metadata JSON")
|
||
|
|
parser.add_argument("--midi", type=str, help="Path to MIDI file")
|
||
|
|
parser.add_argument("--vocal", type=str, help="Path to vocal wav (for midi2meta)")
|
||
|
|
parser.add_argument(
|
||
|
|
"--meta2midi",
|
||
|
|
action="store_true",
|
||
|
|
help="Convert meta -> midi (requires --meta and --midi)",
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
"--midi2meta",
|
||
|
|
action="store_true",
|
||
|
|
help="Convert midi -> meta (requires --midi, --meta, --vocal, --cut_wavs_dir)",
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
"--rmvpe_model_path",
|
||
|
|
type=str,
|
||
|
|
help="Path to RMVPE model",
|
||
|
|
default="pretrained_models/SoulX-Singer-Preprocess/rmvpe/rmvpe.pt",
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
"--device",
|
||
|
|
type=str,
|
||
|
|
help="Device to use for RMVPE",
|
||
|
|
default="cuda",
|
||
|
|
)
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
if args.meta2midi:
|
||
|
|
if not args.meta or not args.midi:
|
||
|
|
parser.error("--meta2midi requires --meta and --midi")
|
||
|
|
meta2midi(args.meta, args.midi)
|
||
|
|
elif args.midi2meta:
|
||
|
|
if not args.midi or not args.meta or not args.vocal:
|
||
|
|
parser.error(
|
||
|
|
"--midi2meta requires --midi, --meta, --vocal"
|
||
|
|
)
|
||
|
|
midi2meta(
|
||
|
|
args.midi,
|
||
|
|
args.meta,
|
||
|
|
args.vocal,
|
||
|
|
rmvpe_model_path=args.rmvpe_model_path,
|
||
|
|
device=args.device,
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
parser.print_help()
|