Module music_df.scripts.dedouble_csvs

We want to process data in the following order: 1. detrill 2. quantize 3. merge notes 4. salami slice 5. dedouble

Functions

def do_csv_file(inputf: str,
config: Config,
output_list: list | multiprocessing.managers.ListProxy)

Expand source code

def do_csv_file(inputf: str, config: Config, output_list: list | ListProxy):
    music_df = read_csv(inputf)
    assert music_df is not None

    if config.quantize:
        music_df = quantize_df(music_df, tpq=config.quantize)
    if config.merge_notes:
        music_df = merge_notes(music_df)

    dedoubled_df = dedouble(music_df)
    relative_path = os.path.relpath(inputf, config.input_folder)
    output_path = os.path.join(config.output_folder, relative_path)
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    dedoubled_df.to_csv(output_path)
    output_list.append(output_path)

def get_csv_and_other_files(input_folder)

Expand source code

def get_csv_and_other_files(input_folder):
    all_files = glob.glob(input_folder + "/**/*", recursive=True)
    csv_files = [file for file in all_files if file.endswith(".csv")]
    other_files = [
        file
        for file in all_files
        if not file.endswith(".csv") and not os.path.isdir(file)
    ]
    return csv_files, other_files

def main()

Expand source code

def main():
    args = parse_args()
    config = Config(**vars(args))

    input_files, other_files = get_csv_and_other_files(config.input_folder)
    if config.max_files is not None:
        input_files = input_files[: config.max_files]

    os.makedirs(config.output_folder, exist_ok=True)
    if config.num_workers > 1:
        manager = Manager()
        output_files = manager.list()
        # error_files = manager.list()
        with Pool(config.num_workers) as pool:
            list(
                tqdm(
                    pool.imap_unordered(
                        partial(
                            do_csv_file,
                            config=config,
                            output_list=output_files,
                        ),
                        input_files,
                    ),
                    total=len(input_files),
                )
            )

    else:
        output_files = []
        # error_files = []
        for inputf in input_files:
            do_csv_file(inputf, config, output_files)

    for file in other_files:
        relative_path = os.path.relpath(file, config.input_folder)
        output_path = os.path.join(config.output_folder, relative_path)
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        shutil.copy2(file, output_path)

def parse_args()

Expand source code

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("input_folder")
    parser.add_argument("output_folder")
    parser.add_argument("--debug", action="store_true")
    parser.add_argument("--num-workers", type=int, default=0)
    parser.add_argument("--max-files", type=int, default=None)
    parser.add_argument("--quantize", type=int, default=None)
    parser.add_argument("--merge-notes", action="store_true")
    args = parser.parse_args()
    return args

Classes

class Config (input_folder: str,
output_folder: str,
debug: bool = False,
num_workers: int = 0,
max_files: int | None = None,
quantize: int | None = None,
merge_notes: bool = False)

Expand source code

@dataclass
class Config:
    input_folder: str
    output_folder: str
    debug: bool = False
    num_workers: int = 0
    max_files: None | int = None
    # 16 is the resolution of musicbert
    quantize: None | int = None
    merge_notes: bool = False

Config(input_folder: str, output_folder: str, debug: bool = False, num_workers: int = 0, max_files: None | int = None, quantize: None | int = None, merge_notes: bool = False)

Instance variables

var debug : bool

var input_folder : str

var max_files : int | None

var merge_notes : bool

var num_workers : int

var output_folder : str

var quantize : int | None