Modify parsing from OHE to real values

src/constants.py

@@ -31,9 +31,9 @@
 # HYPERPARAMETERS #
 
 EPOCHS = 100
-BATCH_SIZE = 8
-MAX_POLYPHONY = 1
-SAMPLE_TIMES = 100
+BATCH_SIZE = 5
+MAX_POLYPHONY = 13
+SAMPLE_TIMES = 1
 
 # Generator
 LR_G = 0.4

src/dataset/preprocessing/parser.py

@@ -1,12 +1,14 @@
 import glob
 import logging
+import re
 from typing import List, Dict
 
 from py_midicsv import midi_to_csv
 from tqdm import tqdm
 
+from constants import RAW_DATASET_PATH, DATASET_PATH, SAMPLE_TIMES, MAX_POLYPHONY
 from dataset.Music import Song, Track, NoteData
-from constants import RAW_DATASET_PATH, DATASET_PATH, NUM_NOTES, MIN_NOTE, SAMPLE_TIMES
+from utils.music import note_to_freq
 
 
 def csv_cleaner(data: List[str]) -> Song:
@@ -27,25 +29,30 @@ def csv_cleaner(data: List[str]) -> Song:
  current_notes: Dict[int, NoteData] = {}
 
  for idx, row in enumerate(data[idx:]):
- row = row.strip().split(', ')
+ row = row.strip()
+
  # Note information events
- if len(row) == 6:
- track, note_time, event, channel, note, velocity = row
- # Note ends event (even if off or no velocity. We add missing end attribute and push to list of data
+ if re.match(r'^\d+, \d+, Note_(on|off)_c, \d+, \d+, \d+$', row):
+ track, note_time, event, channel, note, velocity = row.split(', ')
+
+ # Note ends event (event if off or 0 velocity)
+ # We add missing end attribute and push to list of data
  if event == 'Note_off_c' or (event == 'Note_on_c' and velocity == '0'):
  # TODO: Review strange missing starts
  if note in current_notes:
  current_notes[note].note_end = int(note_time)
  notes_data.append(current_notes[note])
  del current_notes[note]
+
  # Note starts event. Gets start_time, note and velocity information.
  elif event == 'Note_on_c':
  current_notes[note] = NoteData(int(note_time), 0, int(note), int(velocity))
- else:
- # Push and start new track if end of track reached
- if row[2] == 'End_track' and len(notes_data) != 0:
- tracks.append(Track(notes_data))
- notes_data = []
+
+ # Push and start new track if end of track reached
+ elif re.match(r'^\d+, \d+, End_track$', row) and len(notes_data):
+ tracks.append(Track(notes_data))
+ notes_data = []
+ current_notes = {}
 
  return Song(tracks)
 
@@ -66,29 +73,35 @@ def csv_to_series(song: Song) -> List[List[int]]:
  time_idx = 0
  while time_idx < max_time:
  for track_idx, track_time_idx in enumerate(track_time_indices):
- # Continue if track already finished
+ # Skip if track already finished
  if track_time_indices[track_idx] >= song.get_track(track_idx).len_track:
  continue
- note_data = song.get_track(track_idx).get_note_data(track_time_idx)
+
  # Add note to current time step if it's being played
+ note_data = song.get_track(track_idx).get_note_data(track_time_idx)
  if note_data.is_playing(time_idx * SAMPLE_TIMES):
+ # Add as maximum MAX_POLYPHONY notes for each step
+ if len(series_data[time_idx]) >= MAX_POLYPHONY:
+ continue
  series_data[time_idx].append(note_data.note)
+
  # Check if note won't be played on next time step
  if not note_data.is_playing((time_idx + 1) * SAMPLE_TIMES):
  track_time_indices[track_idx] += 1
+
  time_idx += 1
 
  return series_data
 
 
-def series_to_one_hot(series: List[List[int]]):
- one_hot_song = []
+def notes_to_freq(series: List[List[int]]) -> List[List[float]]:
+ notes_freqs = []
  for notes in series:
- one_hot_note = [0] * NUM_NOTES
+ freqs = []
  for note in notes:
- one_hot_note[note - MIN_NOTE] = 1
- one_hot_song.append(one_hot_note)
- return one_hot_song
+ freqs.append(note_to_freq(note))
+ notes_freqs.append(freqs)
+ return notes_freqs
 
 
 if __name__ == '__main__':
@@ -106,10 +119,10 @@ def series_to_one_hot(series: List[List[int]]):
  time_series = list(map(csv_to_series, tqdm(csv_preprocessed, ncols=150)))
 
  logging.info('One-hot encoding notes...')
- one_hot_notes = list(map(series_to_one_hot, tqdm(time_series, ncols=150)))
+ input_notes = list(map(notes_to_freq, tqdm(time_series, ncols=150)))
 
  logging.info('Writing note features ...')
- for path, time_steps in tqdm(zip(files, one_hot_notes), ncols=150):
+ for path, time_steps in tqdm(zip(files, input_notes), ncols=150):
  file = path.split('/')[-1][:-4] + '.txt'
  with open(f'{DATASET_PATH}/{file}', mode='w') as f:
  for ts in time_steps:

src/dataset/preprocessing/reconstructor.py

@@ -1,9 +1,11 @@
 import logging
 import os
 
+import numpy as np
 from py_midicsv import csv_to_midi, FileWriter
 
-from constants import RESULTS_PATH, SAMPLE_TIMES, MIN_NOTE
+from constants import RESULTS_PATH, SAMPLE_TIMES, DATASET_PATH, MAX_POLYPHONY
+from utils.music import freq_to_note
 from utils.typings import NDArray
 
 
@@ -30,32 +32,43 @@ def store_csv_to_midi(title: str, data: str) -> str:
  return f'{RESULTS_PATH}/{title}.mid'
 
 
-def parse_data(data: NDArray) -> str:
+def parse_data(notes_data: NDArray) -> str:
  """
  Parses song data to CSV format with MIDI event format.
- :param data: List of the note played in each time step.
+ :param notes_data: List of the note played in each time step.
  :return: String containing the CSV data of the notes
  """
  logging.info('Parsing note data...')
 
- current_note = None
- csv_data = []
+ start_times = [0] * MAX_POLYPHONY
+ current_notes = [0] * MAX_POLYPHONY
+ csv_data_tracks = [[f'{idx + 2}, 0, Start_track'] for idx in range(MAX_POLYPHONY)]
 
  # TODO: Generalize for polyphony
- start_time = 0
- for time_step, note in enumerate(data):
- if note != current_note:
- if current_note is not None:
- csv_data.append(f'2, {start_time * SAMPLE_TIMES}, Note_on_c, 0, {note + MIN_NOTE}, 64\n'
- f'2, {time_step * SAMPLE_TIMES}, Note_off_c, 0, {note + MIN_NOTE}, 0')
-
- if note != 0:
- start_time = time_step
- current_note = note
- else: # Silence
- current_note = None
-
- return '\n'.join(csv_data)
+ start_times = [0] * MAX_POLYPHONY
+ for time_step, freqs in enumerate(notes_data):
+ notes = list(map(freq_to_note, freqs))
+ for idx, note in enumerate(notes):
+ if note != current_notes[idx]:
+ if current_notes[idx] > 0:
+ channel = idx if idx < 10 else idx + 1
+ csv_data_tracks[idx].append(
+ f'{idx + 2}, {start_times[idx] * SAMPLE_TIMES}, Note_on_c, {channel}, {note}, 64\n' +
+ f'{idx + 2}, {time_step * SAMPLE_TIMES}, Note_off_c, {channel}, {note}, 0'
+ )
+
+ if note != 0:
+ start_times[idx] = time_step
+ current_notes[idx] = note
+ else: # Silence
+ current_notes[idx] = 0
+
+ data_tracks = []
+ for idx in range(len(csv_data_tracks)):
+ csv_data_tracks[idx].append(f'{idx + 2}, 10000, End_track')
+ data_tracks.append('\n'.join(csv_data_tracks[idx]))
+
+ return '\n'.join(data_tracks)
 
 
 def series_to_csv(title: str, data: NDArray) -> str:
@@ -67,17 +80,17 @@ def series_to_csv(title: str, data: NDArray) -> str:
  """
  logging.info('Converting to CSV...')
 
- header = ['0, 0, Header, 1, 2, 480',
+ header = [f'0, 0, Header, 1, {MAX_POLYPHONY}, 384',
  '1, 0, Start_track',
- '1, 0, Title_t, "' + title + '"',
+ f'1, 0, Title_t, "{title}"',
  '1, 0, Time_signature, 4, 2, 24, 8',
- '1, 0, Tempo, 500000',
+ '1, 0, Tempo, 550000',
  '1, 0, End_track',
  '2, 0, Start_track',
  '2, 0, Text_t, "RH"']
  header = '\n'.join(header)
 
- footer = ['2, 4800, End_track', '0, 0, End_of_file']
+ footer = ['0, 0, End_of_file']
  footer = '\n'.join(footer)
 
  return f'{header}\n{parse_data(data)}\n{footer}'

src/utils/music.py

@@ -0,0 +1,24 @@
+import math
+
+
+def note_to_freq(note_idx: int) -> float:
+ """
+ Converts MIDI note index into its corresponding tone frequency.
+ f_n = 2^(n/12)*440 | Reference: https://newt.phys.unsw.edu.au/jw/notes.html
+ :param note_idx: MIDI note index.
+ :return: Note frequency in Hz
+ """
+ return math.pow(2, (note_idx - 69) / 12) * 440
+
+
+def freq_to_note(note_freq: float) -> int:
+ """
+ Converts note frequency into its corresponding MIDI tone index.
+ f_n = 2^(n/12)*440 | Reference: https://newt.phys.unsw.edu.au/jw/notes.html
+ :param note_freq: Note frequency in Hz.
+ :return: MIDI note index.
+ """
+ if note_freq > 0:
+ return int(12 * math.log2(note_freq / 440)) + 69
+ else:
+ return 0