Soul-AILab/SoulX-Singer
185
1"""2SoulX-Singer MIDI <-> metadata converter.3 4Converts between SoulX-Singer-style metadata JSON (with note_text, note_dur,5note_pitch, note_type per segment) and standard MIDI files. Uses an internal6Note dataclass (start_s, note_dur, note_text, note_pitch, note_type) as the7intermediate representation.8"""9import os10import json11import shutil12from dataclasses import dataclass13from typing import Any, List, Tuple, Union14 15import librosa16import mido17from soundfile import write18 19from .f0_extraction import F0Extractor20from .g2p import g2p_transform21 22 23# Audio and segmenting constants (used by _edit_data_to_meta)24SAMPLE_RATE = 4410025DEFAULT_LANGUAGE = "Mandarin"26MAX_GAP_SEC = 5.0 # gap (sec) above which we start a new segment27MAX_SEGMENT_DUR_SUM_SEC = 60.0 # max cumulative note duration per segment (sec)28MIN_GAP_THRESHOLD_SEC = 0.001 # ignore gaps smaller than this29LONG_SILENCE_THRESHOLD_SEC = 0.05 # treat as separate <SP> if gap larger30MAX_LEADING_SP_DUR_SEC = 2.0 # cap leading silence in a segment to this (sec)31DEFAULT_RMVPE_MODEL_PATH = "pretrained_models/SoulX-Singer-Preprocess/rmvpe/rmvpe.pt"32 33 34@dataclass35class Note:36 """Single note: text, duration (seconds), pitch (MIDI), type. start_s is absolute start time in seconds (for ordering / MIDI)."""37 start_s: float38 note_dur: float39 note_text: str40 note_pitch: int41 note_type: int42 43 @property44 def end_s(self) -> float:45 return self.start_s + self.note_dur46 47 48 49def remove_duplicate_segments(meta_data: List[dict]) -> None:50 """Merge consecutive identical notes (same text, pitch, type) within each segment. Mutates meta_data in place."""51 for idx, segment in enumerate(meta_data):52 texts = segment["note_text"]53 durs = segment["note_dur"]54 pitches = segment["note_pitch"]55 types = segment["note_type"]56 new_texts = []57 new_durs = []58 new_pitches = []59 new_types = []60 for i in range(len(texts)):61 if i == 0:62 new_texts.append(texts[i])63 new_durs.append(durs[i])64 new_pitches.append(pitches[i])65 new_types.append(types[i])66 continue67 t, d, p, ty = texts[i], durs[i], pitches[i], types[i]68 if t == "<SP>" and texts[i - 1] == "<SP>":69 new_durs[-1] += d70 continue71 if t == texts[i - 1] and p == pitches[i - 1] and ty == types[i - 1]:72 new_durs[-1] += d73 else:74 new_texts.append(t)75 new_durs.append(d)76 new_pitches.append(p)77 new_types.append(ty)78 meta_data[idx]["note_text"] = new_texts79 meta_data[idx]["note_dur"] = new_durs80 meta_data[idx]["note_pitch"] = new_pitches81 meta_data[idx]["note_type"] = new_types82 83def meta2notes(meta_path: str) -> List[Note]:84 """Parse SoulX-Singer metadata JSON into a flat list of Note (absolute start_s)."""85 with open(meta_path, "r", encoding="utf-8") as f:86 segments = json.load(f)87 if not isinstance(segments, list):88 raise ValueError(f"Metadata must be a list of segments, got {type(segments).__name__}")89 if not segments:90 raise ValueError("Metadata has no segments.")91 92 notes: List[Note] = []93 for seg in segments:94 offset_s = seg["time"][0] / 100095 words = [str(x).replace("<AP>", "<SP>") for i, x in enumerate(seg["text"].split())]96 word_durs = [float(x) for x in seg["duration"].split()]97 pitches = [int(x) for x in seg["note_pitch"].split()]98 types = [int(x) if words[i] != "<SP>" else 1 for i, x in enumerate(seg["note_type"].split())]99 if len(words) != len(word_durs) or len(word_durs) != len(pitches) or len(pitches) != len(types):100 raise ValueError(101 f"Length mismatch in segment {seg.get('item_name', '?')}: "102 "note_text, note_dur, note_pitch, note_type must have same length"103 )104 current_s = offset_s105 for text, dur, pitch, type_ in zip(words, word_durs, pitches, types):106 notes.append(107 Note(108 start_s=current_s,109 note_dur=float(dur),110 note_text=str(text),111 note_pitch=int(pitch),112 note_type=int(type_),113 )114 )115 current_s += float(dur)116 return notes117 118def _append_segment_to_meta(119 meta_path_str: str,120 cut_wavs_output_dir: str,121 vocal_file: str,122 audio_data: Any,123 meta_data: List[dict],124 note_start: List[float],125 note_end: List[float],126 note_text: List[Any],127 note_pitch: List[Any],128 note_type: List[Any],129 note_dur: List[float],130 end_time_ms_override: float | None = None,131) -> None:132 """Write one segment wav and append one segment dict to meta_data. Caller clears note_* lists after."""133 base_name = os.path.splitext(os.path.basename(meta_path_str))[0]134 item_name = f"{base_name}_{len(meta_data)}"135 wav_fn = os.path.join(cut_wavs_output_dir, f"{item_name}.wav")136 start_ms = int(note_start[0] * 1000)137 end_ms = (138 int(end_time_ms_override)139 if end_time_ms_override is not None140 else int(note_end[-1] * 1000)141 )142 start_sample = int(note_start[0] * SAMPLE_RATE)143 end_sample = int(note_end[-1] * SAMPLE_RATE)144 write(wav_fn, audio_data[start_sample:end_sample], SAMPLE_RATE)145 meta_data.append({146 "item_name": item_name,147 "wav_fn": wav_fn,148 "origin_wav_fn": vocal_file,149 "start_time_ms": start_ms,150 "end_time_ms": end_ms,151 "language": DEFAULT_LANGUAGE,152 "note_text": list(note_text),153 "note_pitch": list(note_pitch),154 "note_type": list(note_type),155 "note_dur": list(note_dur),156 })157 158 159def convert_meta(meta_data: List[dict], rmvpe_model_path, device="cuda"):160 pitch_extractor = F0Extractor(rmvpe_model_path, device=device, verbose=False)161 converted_data = []162 163 for item in meta_data:164 wav_fn = item.get("wav_fn")165 if not wav_fn or not os.path.isfile(wav_fn):166 raise FileNotFoundError(f"Segment wav file not found: {wav_fn}")167 f0 = pitch_extractor.process(wav_fn)168 converted_item = {169 "index": item.get("item_name"),170 "language": item.get("language"),171 "time": [item.get("start_time_ms", 0), item.get("end_time_ms", sum(item["note_dur"]) * 1000)],172 "duration": " ".join(str(round(x, 2)) for x in item.get("note_dur", [])),173 "text": " ".join(item.get("note_text", [])),174 "phoneme": " ".join(g2p_transform(item.get("note_text", []), DEFAULT_LANGUAGE)),175 "note_pitch": " ".join(str(x) for x in item.get("note_pitch", [])),176 "note_type": " ".join(str(x) for x in item.get("note_type", [])),177 "f0": " ".join(str(round(float(x), 1)) for x in f0),178 }179 converted_data.append(converted_item)180 181 return converted_data182 183 184def _edit_data_to_meta(185 meta_path_str: str,186 edit_data: List[dict],187 vocal_file: str,188 rmvpe_model_path: str | None = None,189 device: str = "cuda",190) -> None:191 """Write SoulX-Singer metadata JSON from edit_data (list of {start, end, note_text, note_pitch, note_type})."""192 # Use a fixed temporary directory for cut wavs193 cut_wavs_output_dir = os.path.join(os.path.dirname(vocal_file), "cut_wavs_tmp")194 os.makedirs(cut_wavs_output_dir, exist_ok=True)195 196 note_text: List[Any] = []197 note_pitch: List[Any] = []198 note_type: List[Any] = []199 note_dur: List[float] = []200 note_start: List[float] = []201 note_end: List[float] = []202 prev_end = 0.0203 meta_data: List[dict] = []204 audio_data, _ = librosa.load(vocal_file, sr=SAMPLE_RATE, mono=True)205 dur_sum = 0.0206 207 for entry in edit_data:208 start = float(entry["start"])209 end = float(entry["end"])210 text = entry["note_text"]211 pitch = entry["note_pitch"]212 type_ = entry["note_type"]213 214 if text == "" or pitch == "" or type_ == "":215 note_text.append("<SP>")216 note_pitch.append(0)217 note_type.append(1)218 note_dur.append(end - start)219 note_start.append(start)220 note_end.append(end)221 prev_end = end222 dur_sum += end - start223 continue224 225 if (226 len(note_text) > 0227 and note_text[-1] == "<SP>"228 and note_dur[-1] > MAX_LEADING_SP_DUR_SEC229 ):230 cut_time = note_dur[-1] - MAX_LEADING_SP_DUR_SEC231 note_dur[-1] = MAX_LEADING_SP_DUR_SEC232 end_ms_override = note_end[-1] * 1000 - cut_time * 1000233 _append_segment_to_meta(234 meta_path_str,235 cut_wavs_output_dir,236 vocal_file,237 audio_data,238 meta_data,239 note_start,240 note_end,241 note_text,242 note_pitch,243 note_type,244 note_dur,245 end_time_ms_override=end_ms_override,246 )247 note_text = []248 note_pitch = []249 note_type = []250 note_dur = []251 note_start = []252 note_end = []253 prev_end = start254 dur_sum = 0.0255 256 gap_from_prev = start - prev_end257 gap_from_last_note = (start - note_end[-1]) if note_end else 0.0258 if (259 gap_from_prev >= MAX_GAP_SEC260 or gap_from_last_note >= MAX_GAP_SEC261 or dur_sum >= MAX_SEGMENT_DUR_SUM_SEC262 ):263 if len(note_text) > 0:264 _append_segment_to_meta(265 meta_path_str,266 cut_wavs_output_dir,267 vocal_file,268 audio_data,269 meta_data,270 note_start,271 note_end,272 note_text,273 note_pitch,274 note_type,275 note_dur,276 )277 note_text = []278 note_pitch = []279 note_type = []280 note_dur = []281 note_start = []282 note_end = []283 prev_end = start284 dur_sum = 0.0285 286 if start - prev_end > MIN_GAP_THRESHOLD_SEC:287 if start - prev_end > LONG_SILENCE_THRESHOLD_SEC or len(note_text) == 0:288 note_text.append("<SP>")289 note_pitch.append(0)290 note_type.append(1)291 note_dur.append(start - prev_end)292 note_start.append(prev_end)293 note_end.append(start)294 else:295 if len(note_dur) > 0:296 note_dur[-1] += start - prev_end297 note_end[-1] = start298 299 prev_end = end300 note_text.append(text)301 note_pitch.append(int(pitch))302 note_type.append(int(type_))303 note_dur.append(end - start)304 note_start.append(start)305 note_end.append(end)306 dur_sum += end - start307 308 if len(note_text) > 0:309 _append_segment_to_meta(310 meta_path_str,311 cut_wavs_output_dir,312 vocal_file,313 audio_data,314 meta_data,315 note_start,316 note_end,317 note_text,318 note_pitch,319 note_type,320 note_dur,321 )322 323 remove_duplicate_segments(meta_data)324 325 _rmvpe_path = rmvpe_model_path or DEFAULT_RMVPE_MODEL_PATH326 converted_data = convert_meta(meta_data, _rmvpe_path, device)327 328 with open(meta_path_str, "w", encoding="utf-8") as f:329 json.dump(converted_data, f, ensure_ascii=False, indent=2)330 331 # Clean up temporary cut wavs directory332 try:333 shutil.rmtree(cut_wavs_output_dir, ignore_errors=True)334 except Exception:335 pass336 337 338def notes2meta(339 notes: List[Note],340 meta_path: str,341 vocal_file: str,342 rmvpe_model_path: str | None = None,343 device: str = "cuda",344) -> None:345 """Write SoulX-Singer metadata JSON from a list of Note (segmenting + wav cuts)."""346 edit_data = [347 {348 "start": n.start_s,349 "end": n.end_s,350 "note_text": n.note_text,351 "note_pitch": str(n.note_pitch),352 "note_type": str(n.note_type),353 }354 for n in notes355 ]356 _edit_data_to_meta(357 str(meta_path),358 edit_data,359 vocal_file,360 rmvpe_model_path=rmvpe_model_path,361 device=device,362 )363 364 365@dataclass(frozen=True)366class MidiDefaults:367 ticks_per_beat: int = 500368 tempo: int = 500000 # microseconds per beat (120 BPM)369 time_signature: Tuple[int, int] = (4, 4)370 velocity: int = 64371 372 373def _seconds_to_ticks(seconds: float, ticks_per_beat: int, tempo: int) -> int:374 return int(round(seconds * ticks_per_beat * 1_000_000 / tempo))375 376 377def notes2midi(378 notes: List[Note],379 midi_path: str,380 defaults: MidiDefaults | None = None,381) -> None:382 """Write MIDI file from a list of Note."""383 defaults = defaults or MidiDefaults()384 if not notes:385 raise ValueError("Empty note list.")386 387 events: List[Tuple[int, int, Union[mido.Message, mido.MetaMessage]]] = []388 for n in notes:389 start_s = n.start_s390 end_s = n.end_s391 if end_s <= start_s:392 continue393 394 start_ticks = _seconds_to_ticks(395 start_s, defaults.ticks_per_beat, defaults.tempo396 )397 end_ticks = _seconds_to_ticks(398 end_s, defaults.ticks_per_beat, defaults.tempo399 )400 if end_ticks <= start_ticks:401 end_ticks = start_ticks + 1402 403 lyric = n.note_text404 try:405 lyric = lyric.encode("utf-8").decode("latin1")406 except (UnicodeEncodeError, UnicodeDecodeError):407 pass408 if n.note_type == 3:409 lyric = "-"410 411 events.append(412 (start_ticks, 1, mido.MetaMessage("lyrics", text=lyric, time=0))413 )414 events.append(415 (416 start_ticks,417 2,418 mido.Message(419 "note_on",420 note=n.note_pitch,421 velocity=defaults.velocity,422 time=0,423 ),424 )425 )426 events.append(427 (428 end_ticks,429 0,430 mido.Message("note_off", note=n.note_pitch, velocity=0, time=0),431 )432 )433 434 events.sort(key=lambda x: (x[0], x[1]))435 436 mid = mido.MidiFile(ticks_per_beat=defaults.ticks_per_beat)437 track = mido.MidiTrack()438 mid.tracks.append(track)439 440 track.append(mido.MetaMessage("set_tempo", tempo=defaults.tempo, time=0))441 track.append(442 mido.MetaMessage(443 "time_signature",444 numerator=defaults.time_signature[0],445 denominator=defaults.time_signature[1],446 time=0,447 )448 )449 450 last_tick = 0451 for tick, _, msg in events:452 msg.time = max(0, tick - last_tick)453 track.append(msg)454 last_tick = tick455 456 track.append(mido.MetaMessage("end_of_track", time=0))457 mid.save(midi_path)458 459 460def midi2notes(midi_path: str) -> List[Note]:461 """Parse MIDI file into a list of Note. Merges all tracks; tempo from last set_tempo event."""462 mid = mido.MidiFile(midi_path)463 ticks_per_beat = mid.ticks_per_beat464 tempo = 500000465 466 raw_notes: List[dict] = []467 lyrics: List[Tuple[int, str]] = []468 469 for track in mid.tracks:470 abs_ticks = 0471 active = {}472 for msg in track:473 abs_ticks += msg.time474 if msg.type == "set_tempo":475 tempo = msg.tempo476 elif msg.type == "lyrics":477 text = msg.text478 try:479 text = text.encode("latin1").decode("utf-8")480 except Exception:481 pass482 lyrics.append((abs_ticks, text))483 elif msg.type == "note_on":484 key = (msg.channel, msg.note)485 if msg.velocity > 0:486 active[key] = (abs_ticks, msg.velocity)487 else:488 if key in active:489 start_ticks, vel = active.pop(key)490 raw_notes.append(491 {492 "midi": msg.note,493 "start_ticks": start_ticks,494 "duration_ticks": abs_ticks - start_ticks,495 "velocity": vel,496 "lyric": "",497 }498 )499 elif msg.type == "note_off":500 key = (msg.channel, msg.note)501 if key in active:502 start_ticks, vel = active.pop(key)503 raw_notes.append(504 {505 "midi": msg.note,506 "start_ticks": start_ticks,507 "duration_ticks": abs_ticks - start_ticks,508 "velocity": vel,509 "lyric": "",510 }511 )512 513 if not raw_notes:514 raise ValueError("No notes found in MIDI file")515 516 for n in raw_notes:517 n["end_ticks"] = n["start_ticks"] + n["duration_ticks"]518 519 raw_notes.sort(key=lambda n: n["start_ticks"])520 lyrics.sort(key=lambda x: x[0])521 522 trimmed = []523 for note in raw_notes:524 while trimmed:525 prev = trimmed[-1]526 if note["start_ticks"] < prev["end_ticks"]:527 prev["end_ticks"] = note["start_ticks"]528 prev["duration_ticks"] = prev["end_ticks"] - prev["start_ticks"]529 if prev["duration_ticks"] <= 0:530 trimmed.pop()531 continue532 break533 trimmed.append(note)534 raw_notes = trimmed535 536 tolerance = ticks_per_beat // 100537 lyric_idx = 0538 for note in raw_notes:539 while lyric_idx < len(lyrics) and lyrics[lyric_idx][0] < note["start_ticks"] - tolerance:540 lyric_idx += 1541 if lyric_idx < len(lyrics):542 lyric_ticks, lyric_text = lyrics[lyric_idx]543 if abs(lyric_ticks - note["start_ticks"]) <= tolerance:544 note["lyric"] = lyric_text545 lyric_idx += 1546 547 def ticks_to_seconds(ticks: int) -> float:548 return (ticks / ticks_per_beat) * (tempo / 1_000_000)549 550 result: List[Note] = []551 prev_end_s = 0.0552 for idx, n in enumerate(raw_notes):553 start_s = ticks_to_seconds(n["start_ticks"])554 end_s = ticks_to_seconds(n["end_ticks"])555 if prev_end_s > start_s:556 start_s = prev_end_s557 dur_s = end_s - start_s558 if dur_s <= 0:559 continue560 561 lyric = n.get("lyric", "")562 if not lyric:563 tp = 2564 text = "啦"565 elif lyric == "<SP>":566 tp = 1567 text = "<SP>"568 elif lyric == "-":569 tp = 3570 text = raw_notes[idx - 1].get("lyric", "-") if idx > 0 else "-"571 else:572 tp = 2573 text = lyric574 575 result.append(576 Note(577 start_s=start_s,578 note_dur=dur_s,579 note_text=text,580 note_pitch=n["midi"],581 note_type=tp,582 )583 )584 prev_end_s = end_s585 586 return result587 588 589def meta2midi(meta_path: str, midi_path: str, defaults: MidiDefaults | None = None) -> None:590 """Convert SoulX-Singer metadata JSON to MIDI file (meta -> List[Note] -> midi)."""591 notes = meta2notes(meta_path)592 notes2midi(notes, midi_path, defaults)593 print(f"Saved MIDI to {midi_path}")594 595 596def midi2meta(597 midi_path: str,598 meta_path: str,599 vocal_file: str,600 rmvpe_model_path: str | None = None,601 device: str = "cuda",602) -> None:603 """Convert MIDI file to SoulX-Singer metadata JSON (midi -> List[Note] -> meta)."""604 meta_dir = os.path.dirname(meta_path)605 if meta_dir:606 os.makedirs(meta_dir, exist_ok=True)607 # cut_wavs will be written to a fixed temporary directory inside _edit_data_to_meta608 notes = midi2notes(midi_path)609 notes2meta(610 notes,611 meta_path,612 vocal_file,613 rmvpe_model_path=rmvpe_model_path,614 device=device,615 )616 print(f"Saved Meta to {meta_path}")617 618 619if __name__ == "__main__":620 import argparse621 622 parser = argparse.ArgumentParser(623 description="Convert SoulX-Singer metadata JSON <-> MIDI."624 )625 parser.add_argument("--meta", type=str, help="Path to metadata JSON")626 parser.add_argument("--midi", type=str, help="Path to MIDI file")627 parser.add_argument("--vocal", type=str, help="Path to vocal wav (for midi2meta)")628 parser.add_argument(629 "--meta2midi",630 action="store_true",631 help="Convert meta -> midi (requires --meta and --midi)",632 )633 parser.add_argument(634 "--midi2meta",635 action="store_true",636 help="Convert midi -> meta (requires --midi, --meta, --vocal, --cut_wavs_dir)",637 )638 parser.add_argument(639 "--rmvpe_model_path",640 type=str,641 help="Path to RMVPE model",642 default="pretrained_models/SoulX-Singer-Preprocess/rmvpe/rmvpe.pt",643 )644 parser.add_argument(645 "--device",646 type=str,647 help="Device to use for RMVPE",648 default="cuda",649 )650 args = parser.parse_args()651 652 if args.meta2midi:653 if not args.meta or not args.midi:654 parser.error("--meta2midi requires --meta and --midi")655 meta2midi(args.meta, args.midi)656 elif args.midi2meta:657 if not args.midi or not args.meta or not args.vocal:658 parser.error(659 "--midi2meta requires --midi, --meta, --vocal"660 )661 midi2meta(662 args.midi,663 args.meta,664 args.vocal,665 rmvpe_model_path=args.rmvpe_model_path,666 device=args.device,667 )668 else:669 parser.print_help()