# Copyright 2024 The YourMT3 Authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Please see the details in the LICENSE file.
"""event2note.py:
Event to NoteEvent:
• event2note_event
NoteEvent to Note:
• note_event2note
• merge_zipped_note_events_and_ties_to_notes
"""
import warnings
from collections import Counter
from typing import List, Tuple, Optional, Dict, Counter
from utils.note_event_dataclasses import Note, NoteEvent
from utils.note_event_dataclasses import Event
from utils.note2event import validate_notes, trim_overlapping_notes
MINIMUM_OFFSET_SEC = 0.01
DECODING_ERR_TYPES = [
'decoding_time', 'Err/Missing prg in tie', 'Err/Missing tie', 'Err/Shift out of range', 'Err/Missing prg',
'Err/Missing vel', 'Err/Multi-tie type 1', 'Err/Multi-tie type 2', 'Err/Unknown event', 'Err/onset not found',
'Err/active ne incomplete', 'Err/merging segment tie', 'Err/long note > 10s'
]
def event2note_event(events: List[Event],
start_time: float = 0.0,
sort: bool = True,
tps: int = 100) -> Tuple[List[NoteEvent], List[NoteEvent], List[Tuple[int]], Counter[str]]:
"""Convert events to note events.
Args:
events: A list of events.
start_time: The start time of the segment.
sort: Whether to sort the note events.
tps: Ticks per second.
Returns:
List[NoteEvent]: A list of note events.
List[NoteEvent]: A list of tie note events.
List[Tuple[int]]: A list of last activity of segment. [(program, pitch), ...]. This is useful
for validating notes within a batch of segments extracted from a file.
Counter[str]: A dictionary of error counters.
"""
assert (start_time >= 0.)
# Collect tie events
tie_index = program_state = None
tie_note_events = []
last_activity = [] # For activity check and last activity of segment. [(program, pitch), ...]
error_counter = {} # Add a dictionary to count the errors by their types
for i, e in enumerate(events):
try:
if e.type == 'tie':
tie_index = i
break
if e.type == 'shift':
break
elif e.type == 'program':
program_state = e.value
elif e.type == 'pitch':
if program_state is None:
raise ValueError('Err/Missing prg in tie')
tie_note_events.append(
NoteEvent(is_drum=False, program=program_state, time=None, velocity=1, pitch=e.value))
last_activity.append((program_state, e.value)) # (program, pitch)
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
try:
if tie_index is None:
raise ValueError('Err/Missing tie')
else:
events = events[tie_index + 1:]
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
return [], [], [], error_counter
# Collect main events:
note_events = []
velocity_state = None
start_tick = round(start_time * tps)
tick_state = start_tick
# keep the program_state of last tie event...
for e in events:
try:
if e.type == 'shift':
if e.value <= 0 or e.value > 1000:
raise ValueError('Err/Shift out of range')
# tick_state += e.value
tick_state = start_tick + e.value
elif e.type == 'drum':
note_events.append(
NoteEvent(is_drum=True, program=128, time=tick_state / tps, velocity=1, pitch=e.value))
elif e.type == 'program':
program_state = e.value
elif e.type == 'velocity':
velocity_state = e.value
elif e.type == 'pitch':
if program_state is None:
raise ValueError('Err/Missing prg')
elif velocity_state is None:
raise ValueError('Err/Missing vel')
# Check activity
if velocity_state > 0:
last_activity.append((program_state, e.value)) # (program, pitch)
elif velocity_state == 0 and (program_state, e.value) in last_activity:
last_activity.remove((program_state, e.value))
else:
# print(f'tick_state: {tick_state}') # <-- This displays unresolved offset errors!!
raise ValueError('Err/Note off without note on')
note_events.append(
NoteEvent(is_drum=False,
program=program_state,
time=tick_state / tps,
velocity=velocity_state,
pitch=e.value))
elif e.type == 'EOS':
break
elif e.type == 'PAD':
continue
elif e.type == 'UNK':
continue
elif e.type == 'tie':
if tick_state == start_tick:
raise ValueError('Err/Multi-tie type 1')
else:
raise ValueError('Err/Multi-tie type 2')
else:
raise ValueError(f'Err/Unknown event')
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
if sort:
note_events.sort(key=lambda n_ev: (n_ev.time, n_ev.is_drum, n_ev.program, n_ev.velocity, n_ev.pitch))
tie_note_events.sort(key=lambda n_ev: (n_ev.is_drum, n_ev.program, n_ev.pitch))
return note_events, tie_note_events, last_activity, error_counter
def note_event2note(
note_events: List[NoteEvent],
tie_note_events: Optional[List[NoteEvent]] = None,
sort: bool = True,
fix_offset: bool = True,
trim_overlap: bool = True,
) -> Tuple[List[Note], Counter[str]]:
"""Convert note events to notes.
Returns:
List[Note]: A list of merged note events.
Counter[str]: A dictionary of error counters.
"""
notes = []
active_note_events = {}
error_counter = {} # Add a dictionary to count the errors by their types
if tie_note_events is not None:
for ne in tie_note_events:
active_note_events[(ne.pitch, ne.program)] = ne
if sort:
note_events.sort(key=lambda ne: (ne.time, ne.is_drum, ne.pitch, ne.velocity, ne.program))
for ne in note_events:
try:
if ne.time == None:
continue
elif ne.is_drum:
if ne.velocity == 1:
notes.append(
Note(is_drum=True,
program=128,
onset=ne.time,
offset=ne.time + MINIMUM_OFFSET_SEC,
pitch=ne.pitch,
velocity=1))
else:
continue
elif ne.velocity == 1:
active_ne = active_note_events.get((ne.pitch, ne.program))
if active_ne is not None:
active_note_events.pop((ne.pitch, ne.program))
notes.append(
Note(False, active_ne.program, active_ne.time, ne.time, active_ne.pitch, active_ne.velocity))
active_note_events[(ne.pitch, ne.program)] = ne
elif ne.velocity == 0:
active_ne = active_note_events.pop((ne.pitch, ne.program), None)
if active_ne is not None:
notes.append(
Note(False, active_ne.program, active_ne.time, ne.time, active_ne.pitch, active_ne.velocity))
else:
raise ValueError('Err/onset not found')
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
for ne in active_note_events.values():
try:
if ne.velocity == 1:
if ne.program == None or ne.pitch == None:
raise ValueError('Err/active ne incomplete')
elif ne.time == None:
continue
else:
notes.append(
Note(is_drum=False,
program=ne.program,
onset=ne.time,
offset=ne.time + MINIMUM_OFFSET_SEC,
pitch=ne.pitch,
velocity=1))
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
if fix_offset:
for n in list(notes):
try:
if n.offset - n.onset > 10:
n.offset = n.onset + MINIMUM_OFFSET_SEC
raise ValueError('Err/long note > 10s')
except ValueError as ve:
error_type = str(ve)
error_counter[error_type] = error_counter.get(error_type, 0.) + 1
if sort:
notes.sort(key=lambda note: (note.onset, note.is_drum, note.program, note.velocity, note.pitch))
if fix_offset:
notes = validate_notes(notes, fix=True)
if trim_overlap:
notes = trim_overlapping_notes(notes, sort=True)
return notes, error_counter
def merge_zipped_note_events_and_ties_to_notes(zipped_note_events_and_ties,
force_note_off_missing_tie=True,
fix_offset=True) -> Tuple[List[Note], Counter[str]]:
"""Merge zipped note events and ties.
Args:
zipped_note_events_and_ties: A list of tuples of (note events, tie note events, last_activity, start time).
force_note_off_missing_tie: Whether to force note off for missing tie note events.
fix_offset: Whether to fix the offset of notes.
Returns:
List[Note]: A list of merged note events.
Counter[str]: A dictionary of error counters.
"""
merged_note_events = []
prev_last_activity = None
seg_merge_err_cnt = Counter()
for nes, tie_nes, last_activity, start_time in zipped_note_events_and_ties:
if prev_last_activity is not None and force_note_off_missing_tie:
# Check mismatch between prev_last_activity and current tie_note_events
prog_pitch_tie = set([(ne.program, ne.pitch) for ne in tie_nes])
for prog_pitch_pla in prev_last_activity: # (program, pitch) of previous last active notes
if prog_pitch_pla not in prog_pitch_tie:
# last acitve notes of previous segment is missing in tie information.
# We create a note off event for these notes at the beginning of current note events.
merged_note_events.append(
NoteEvent(is_drum=False,
program=prog_pitch_pla[0],
time=start_time,
velocity=0,
pitch=prog_pitch_pla[1]))
seg_merge_err_cnt['Err/merging segment tie'] += 1
else:
pass
merged_note_events += nes
prev_last_activity = last_activity
# merged_note_events to notes
notes, err_cnt = note_event2note(merged_note_events, tie_note_events=None, fix_offset=fix_offset)
# gather error counts
err_cnt.update(seg_merge_err_cnt)
return notes, err_cnt