|
|
import json |
|
|
import math |
|
|
import warnings |
|
|
from collections import OrderedDict |
|
|
|
|
|
import librosa |
|
|
import numpy as np |
|
|
import tqdm |
|
|
import pathlib |
|
|
from csv import DictReader, DictWriter |
|
|
|
|
|
import click |
|
|
|
|
|
from get_pitch import get_pitch_parselmouth |
|
|
|
|
|
warns = [] |
|
|
|
|
|
|
|
|
def get_aligned_pitch(wav_path: pathlib.Path, total_secs: float, timestep: float): |
|
|
waveform, _ = librosa.load(wav_path, sr=44100, mono=True) |
|
|
_, f0, _ = get_pitch_parselmouth(waveform, 512, 44100) |
|
|
pitch = librosa.hz_to_midi(f0) |
|
|
if pitch.shape[0] < total_secs / timestep: |
|
|
pad = math.ceil(total_secs / timestep) - pitch.shape[0] |
|
|
pitch = np.pad(pitch, [0, pad], mode='constant', constant_values=[0, pitch[-1]]) |
|
|
return pitch |
|
|
|
|
|
|
|
|
def correct_cents_item( |
|
|
name: str, item: OrderedDict, ref_pitch: np.ndarray, |
|
|
timestep: float, error_ratio: float |
|
|
): |
|
|
note_seq = item['note_seq'].split() |
|
|
note_dur = [float(d) for d in item['note_dur'].split()] |
|
|
assert len(note_seq) == len(note_dur) |
|
|
|
|
|
start = 0. |
|
|
note_seq_correct = [] |
|
|
for i, (note, dur) in enumerate(zip(note_seq, note_dur)): |
|
|
end = start + dur |
|
|
if note == 'rest': |
|
|
start = end |
|
|
note_seq_correct.append('rest') |
|
|
continue |
|
|
|
|
|
midi = librosa.note_to_midi(note, round_midi=False) |
|
|
start_idx = math.floor(start / timestep) |
|
|
end_idx = math.ceil(end / timestep) |
|
|
note_pitch = ref_pitch[start_idx: end_idx] |
|
|
note_pitch_close = note_pitch[(note_pitch >= midi - 0.5) & (note_pitch < midi + 0.5)] |
|
|
if len(note_pitch_close) < len(note_pitch) * error_ratio or len(note_pitch) == 0: |
|
|
warns.append({ |
|
|
'position': name, |
|
|
'note_index': i, |
|
|
'note_value': note |
|
|
}) |
|
|
if len(note_pitch) == 0 or len(note_pitch_close) == 0: |
|
|
start = end |
|
|
note_seq_correct.append(note) |
|
|
continue |
|
|
midi_correct = np.mean(note_pitch_close) |
|
|
note_seq_correct.append(librosa.midi_to_note(midi_correct, cents=True, unicode=False)) |
|
|
|
|
|
start = end |
|
|
|
|
|
item['note_seq'] = ' '.join(note_seq_correct) |
|
|
|
|
|
|
|
|
def save_warnings(save_dir: pathlib.Path): |
|
|
if len(warns) > 0: |
|
|
save_path = save_dir.resolve() / 'warnings.csv' |
|
|
with open(save_path, 'w', encoding='utf8', newline='') as f: |
|
|
writer = DictWriter(f, fieldnames=['position', 'note_index', 'note_value']) |
|
|
writer.writeheader() |
|
|
writer.writerows(warns) |
|
|
warnings.warn( |
|
|
message=f'possible labeling errors saved in {save_path}', |
|
|
category=UserWarning |
|
|
) |
|
|
warnings.filterwarnings(action='default') |
|
|
|
|
|
|
|
|
@click.group(help='Apply cents correction to note sequences') |
|
|
def correct_cents(): |
|
|
pass |
|
|
|
|
|
|
|
|
@correct_cents.command(help='Apply cents correction to note sequences in transcriptions.csv') |
|
|
@click.argument('transcriptions', metavar='TRANSCRIPTIONS') |
|
|
@click.argument('waveforms', metavar='WAVS') |
|
|
@click.option('--error_ratio', metavar='RATIO', type=float, default=0.4, |
|
|
help='If the percentage of pitch points within a deviation of 50 cents compared to the note label ' |
|
|
'is lower than this value, a warning will be raised.') |
|
|
def csv( |
|
|
transcriptions, |
|
|
waveforms, |
|
|
error_ratio |
|
|
): |
|
|
transcriptions = pathlib.Path(transcriptions).resolve() |
|
|
waveforms = pathlib.Path(waveforms).resolve() |
|
|
with open(transcriptions, 'r', encoding='utf8') as f: |
|
|
reader = DictReader(f) |
|
|
items: list[OrderedDict] = [] |
|
|
for item in reader: |
|
|
items.append(OrderedDict(item)) |
|
|
|
|
|
timestep = 512 / 44100 |
|
|
for item in tqdm.tqdm(items): |
|
|
item: OrderedDict |
|
|
ref_pitch = get_aligned_pitch( |
|
|
wav_path=waveforms / (item['name'] + '.wav'), |
|
|
total_secs=sum(float(d) for d in item['note_dur'].split()), |
|
|
timestep=timestep |
|
|
) |
|
|
correct_cents_item( |
|
|
name=item['name'], item=item, ref_pitch=ref_pitch, |
|
|
timestep=timestep, error_ratio=error_ratio |
|
|
) |
|
|
|
|
|
with open(transcriptions, 'w', encoding='utf8', newline='') as f: |
|
|
writer = DictWriter(f, fieldnames=['name', 'ph_seq', 'ph_dur', 'ph_num', 'note_seq', 'note_dur']) |
|
|
writer.writeheader() |
|
|
writer.writerows(items) |
|
|
save_warnings(transcriptions.parent) |
|
|
|
|
|
|
|
|
@correct_cents.command(help='Apply cents correction to note sequences in DS files') |
|
|
@click.argument('ds_dir', metavar='DS_DIR') |
|
|
@click.option('--error_ratio', metavar='RATIO', type=float, default=0.4, |
|
|
help='If the percentage of pitch points within a deviation of 50 cents compared to the note label ' |
|
|
'is lower than this value, a warning will be raised.') |
|
|
def ds( |
|
|
ds_dir, |
|
|
error_ratio |
|
|
): |
|
|
ds_dir = pathlib.Path(ds_dir).resolve() |
|
|
assert ds_dir.exists(), 'The directory of DS files does not exist.' |
|
|
|
|
|
timestep = 512 / 44100 |
|
|
for ds_file in tqdm.tqdm(ds_dir.glob('*.ds')): |
|
|
if not ds_file.is_file(): |
|
|
continue |
|
|
|
|
|
assert ds_file.with_suffix('.wav').exists(), \ |
|
|
f'Missing corresponding .wav file of {ds_file.name}.' |
|
|
with open(ds_file, 'r', encoding='utf8') as f: |
|
|
params = json.load(f) |
|
|
if not isinstance(params, list): |
|
|
params = [params] |
|
|
params = [OrderedDict(p) for p in params] |
|
|
|
|
|
ref_pitch = get_aligned_pitch( |
|
|
wav_path=ds_file.with_suffix('.wav'), |
|
|
total_secs=params[-1]['offset'] + sum(float(d) for d in params[-1]['note_dur'].split()), |
|
|
timestep=timestep |
|
|
) |
|
|
for i, param in enumerate(params): |
|
|
start_idx = math.floor(param['offset'] / timestep) |
|
|
end_idx = math.ceil((param['offset'] + sum(float(d) for d in param['note_dur'].split())) / timestep) |
|
|
correct_cents_item( |
|
|
name=f'{ds_file.stem}#{i}', item=param, ref_pitch=ref_pitch[start_idx: end_idx], |
|
|
timestep=timestep, error_ratio=error_ratio |
|
|
) |
|
|
|
|
|
with open(ds_file, 'w', encoding='utf8') as f: |
|
|
json.dump(params, f, ensure_ascii=False, indent=2) |
|
|
save_warnings(ds_dir) |
|
|
|
|
|
|
|
|
if __name__ == '__main__': |
|
|
correct_cents() |
|
|
|
