Muse Dash Soundtracking Encoding Script

encode_music.py

#!/usr/bin/python3
"""Muse Dash Soundtrack Encoding Script

Prerequisites:
    - `opus-tools` installed (this script uses `opusenc`)
    - Muse Dash tracks extracted as WAVs to a directory
    - Muse Dash cover images extracted as PNGs to a directory
    - CSV file describing each track in Muse Dash

See the blog post for details:
    https://deltaepsilon.ca/posts/muse-dash-soundtrack/
"""

import csv
import glob
import multiprocessing
import os
import queue
import subprocess
import threading

NUM_THREADS = multiprocessing.cpu_count()
OUT_DIR = "output"
AUDIOCLIP_DIR = "AudioClip"
TEXTURE2D_DIR = "Texture2D"
CSVFILE = "muse_dash_songs.csv"
ENCCMD = [
    'opusenc',
    '--bitrate', '128',
    '--title', '"{title}"',
    '--artist', '"{artist}"',
    '--album', '"Muse Dash - {pack}"',
    '--genre', '"Video Games"',
    '--picture', '{pic}',
    '--comment', 'tracknumber={tracknumber}',
    '--comment', 'tracktotal={tracktotal}',
    '{infile}',
    '{outfile}'
]

def encode_thread(rq, wavfiles, artfiles):
    row = rq.get()
    while row is not None:
        if "music2" not in row["Song filename"]:
            wav = row["Song filename"] + "_music.wav"
            art = row["Song filename"] + "_cover.png"
            out = row["Song filename"] + ".opus"
        else:
            wav = row["Song filename"] + ".wav"
            art = row["Song filename"][:-len("_music2")] + "_cover.png"
            out = row["Song filename"][:-len("_music2")] + "2.opus"
        wav = os.path.join(AUDIOCLIP_DIR, wav)
        art = os.path.join(TEXTURE2D_DIR, art)
        out = os.path.join(OUT_DIR, out)
        if wav not in wavfiles:
            print("Missing WAV: ", wav)
            continue
        if art not in artfiles:
            print("Missing cover art: ", art)
            continue
        formatted_cmd = []
        for part in ENCCMD:
            formatted_cmd.append(part.format(
                title=row["Song Name"].strip(),
                artist=row["Artist"].strip(),
                pack=row["Pack"].strip(),
                pic=art,
                tracknumber=row["Track Number"],
                tracktotal=row["Track Total"],
                infile=wav,
                outfile=out
            ))
        subprocess.run(formatted_cmd, check=True)
        #print(formatted_cmd)
        row = rq.get()

def main():
    os.makedirs(OUT_DIR, exist_ok=True)
    wavfiles = set(glob.glob(os.path.join(AUDIOCLIP_DIR, "*.wav")))
    artfiles = set(glob.glob(os.path.join(TEXTURE2D_DIR, "*.png")))
    rq = queue.Queue()
    threads = []
    for _ in range(NUM_THREADS):
        thread = threading.Thread(target=encode_thread, args=(rq, wavfiles, artfiles))
        thread.start()
        threads.append(thread)
    with open(CSVFILE, newline="") as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            rq.put(row)
            
    for thread in threads:
        rq.put(None)
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    main()