Skip to content

Instantly share code, notes, and snippets.

@timonwong
Created September 15, 2013 11:48
Show Gist options
  • Save timonwong/6570051 to your computer and use it in GitHub Desktop.
Save timonwong/6570051 to your computer and use it in GitHub Desktop.
Batch convert to .spx (through ffmpeg)
# -*- coding: utf-8 -*-
import os
import subprocess
import logging
import multiprocessing
import shutil
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
# -- Configuration --
# Path to ffmpeg binary
FFMPEG_BIN = r'ffmpeg.exe'
# Path to data folder (source directory)
# wav files will be converted to spx
# other files will just be copyied to destination directory
SRC_DIR = r'<BLAHBLAHBLAH>'
# Destination directory
DST_DIR = r'<BLAHBLAHBLAH>'
# -- Constatns --
AUDIO_FILE_TYPES = set(['.mp3', '.wav', '.ogg', '.m4a'])
def converter_process(q):
while True:
item = q.get()
if item is None:
break
wav_fullname, spx_fullname = item
basename = os.path.basename(wav_fullname)
status = subprocess.call([
FFMPEG_BIN, '-loglevel', 'error', '-y',
'-i', wav_fullname,
'-map_metadata', '-1',
'-acodec', 'speex',
'-qscale:a', '8',
spx_fullname
])
if status != 0:
logging.error('Convert "%s" failed', basename)
q.task_done()
logging.info('Process %d finished', multiprocessing.current_process().pid)
q.task_done()
def main():
# No color for ffmpeg
os.environ['AV_LOG_FORCE_NOCOLOR'] = '1'
task_queue = multiprocessing.JoinableQueue(100)
num_processes = multiprocessing.cpu_count()
processes = []
for i in range(num_processes):
p = multiprocessing.Process(target=converter_process,
args=(task_queue, ))
p.daemon = True
p.start()
processes.append(p)
logging.info('Started %d processes', num_processes)
for directory, dirnames, filenames in os.walk(SRC_DIR):
n = 0
count = len(filenames)
for filename in filenames:
n += 1
rel_dir = os.path.relpath(directory, SRC_DIR)
base, ext = os.path.splitext(filename)
src_fullname = os.path.normpath(
os.path.join(SRC_DIR, rel_dir, filename))
dst_fullname = os.path.normpath(
os.path.join(DST_DIR, rel_dir, filename))
if (n % 100) == 0:
logging.info(
'Progress: %.1f%% (%d/%d)', float(n) / count * 100, n, count
)
# Copy files that are not audio files
if ext.lower() not in AUDIO_FILE_TYPES:
if not os.path.exists(dst_fullname):
shutil.copy(src_fullname, dst_fullname)
continue
spx_fullname = os.path.join(DST_DIR, base + '.spx')
# Check spx converted
if os.path.exists(spx_fullname):
continue
task_queue.put((src_fullname, spx_fullname))
task_queue.put(None)
task_queue.put(None)
task_queue.join()
logging.info('All done')
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment