Congegni/sintonia
7
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

CC-430: Audio normalization (Replaygain Support)

Browse source
This commit is contained in:
Martin Konecny 2012-07-12 17:58:29 -04:00
parent 62287a2313
commit f0f033b4fb
7 changed files with 401 additions and 357 deletions
Download patch file Download diff file Expand all files Collapse all files

88
python_apps/media-monitor/airtimefilemonitor/replaygain.py
View file

@ -2,6 +2,8 @@ from subprocess import Popen, PIPE
import re
import os
import sys
import shutil
import tempfile
def get_process_output(command):
"""
@ -26,45 +28,69 @@ def get_mime_type(file_path):
return get_process_output("timeout 5 file -b --mime-type %s" % file_path)
def duplicate_file(file_path):
"""
Makes a duplicate of the file and returns the path of this duplicate file.
"""
fsrc = open(file_path, 'r')
fdst = tempfile.NamedTemporaryFile(delete=False)
print "Copying %s to %s" % (file_path, fdst.name)
shutil.copyfileobj(fsrc, fdst)
fsrc.close()
fdst.close()
return fdst.name
def calculate_replay_gain(file_path):
"""
This function accepts files of type mp3/ogg/flac and returns a calculated ReplayGain value in dB.
If the value cannot be calculated for some reason, then we default to 0 (Unity Gain).
TODO:
Currently some of the subprocesses called will actually insert metadata into the file itself,
which we do *not* want as this changes the file's hash. Need to make a copy of the file before
we run this function.
http://wiki.hydrogenaudio.org/index.php?title=ReplayGain_1.0_specification
"""
search = None
if re.search(r'mp3$', file_path, re.IGNORECASE) or get_mime_type(file_path) == "audio/mpeg":
if run_process("which mp3gain > /dev/null") == 0:
out = get_process_output('mp3gain -q "%s" 2> /dev/null' % file_path)
search = re.search(r'Recommended "Track" dB change: (.*)', out)
try:
"""
Making a duplicate is required because the ReplayGain extraction utilities we use
make unwanted modifications to the file.
"""
search = None
temp_file_path = duplicate_file(file_path)
if re.search(r'mp3$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "audio/mpeg":
if run_process("which mp3gain > /dev/null") == 0:
out = get_process_output('mp3gain -q "%s" 2> /dev/null' % temp_file_path)
search = re.search(r'Recommended "Track" dB change: (.*)', out)
else:
print "mp3gain not found"
#Log warning
elif re.search(r'ogg$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "application/ogg":
if run_process("which vorbisgain > /dev/null && which ogginfo > /dev/null") == 0:
run_process('vorbisgain -q -f "%s" 2>/dev/null >/dev/null' % temp_file_path)
out = get_process_output('ogginfo "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "vorbisgain/ogginfo not found"
#Log warning
elif re.search(r'flac$', temp_file_path, re.IGNORECASE) or get_mime_type(temp_file_path) == "audio/x-flac":
if run_process("which metaflac > /dev/null") == 0:
out = get_process_output('metaflac --show-tag=REPLAYGAIN_TRACK_GAIN "%s"' % temp_file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "metaflac not found"
#Log warning
else:
print "mp3gain not found"
#Log warning
elif re.search(r'ogg$', file_path, re.IGNORECASE) or get_mime_type(file_path) == "application/ogg":
if run_process("which vorbisgain > /dev/null && which ogginfo > /dev/null") == 0:
run_process('vorbisgain -q -f "%s" 2>/dev/null >/dev/null' % file_path)
out = get_process_output('ogginfo "%s"' % file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "vorbisgain/ogginfo not found"
#Log warning
elif re.search(r'flac$', file_path, re.IGNORECASE) or get_mime_type(file_path) == "audio/x-flac":
if run_process("which metaflac > /dev/null") == 0:
out = get_process_output('metaflac --show-tag=REPLAYGAIN_TRACK_GAIN "%s"' % file_path)
search = re.search(r'REPLAYGAIN_TRACK_GAIN=(.*) dB', out)
else:
print "metaflac not found"
#Log warning
else:
pass
#Log unknown file type.
pass
#Log unknown file type.
#no longer need the temp, file simply remove it.
os.remove(temp_file_path)
except Exception, e:
print e
replay_gain = 0
if search:

78
python_apps/media-monitor/airtimefilemonitor/replaygainupdater.py Normal file
View file

@ -0,0 +1,78 @@
from threading import Thread
import traceback
import os
import logging
import json
from api_clients import api_client
import replaygain
class ReplayGainUpdater(Thread):
"""
The purpose of the class is to query the server for a list of files which do not have a ReplayGain
value calculated. This class will iterate over the list calculate the values, update the server and
repeat the process until the the server reports there are no files left.
This class will see heavy activity right after a 2.1->2.2 upgrade since 2.2 introduces ReplayGain
normalization. A fresh install of Airtime 2.2 will see this class not used at all since a file
imported in 2.2 will automatically have its ReplayGain value calculated.
"""
def __init__(self, logger):
Thread.__init__(self)
self.logger = logger
self.api_client = api_client.AirTimeApiClient()
def main(self):
#TODO
directories = self.api_client.list_all_watched_dirs()['dirs']
for dir_id, dir_path in directories.iteritems():
try:
processed_data = []
#keep getting 100 rows at a time for current music_dir (stor or watched folder).
#When we get a response with 0 rows, then we will set response to True.
finished = False
while not finished:
# return a list of pairs where the first value is the file's database row id
# and the second value is the filepath
file_path = self.api_client.get_files_without_replay_gain_value(dir_id)
print "temp file saved to %s" % file_path
num_lines = 0
with open(file_path) as f:
for line in f:
num_lines += 1
data = json.loads(line.strip())
track_path = os.path.join(dir_path, data['fp'])
processed_data.append((data['id'], replaygain.calculate_replay_gain(track_path)))
if num_lines == 0:
finished = True
os.remove(file_path)
#send data here
pass
except Exception, e:
print e
def run(self):
try: self.main()
except Exception, e:
self.logger.error('ReplayGainUpdater Exception: %s', traceback.format_exc())
self.logger.error(e)
if __name__ == "__main__":
try:
rgu = ReplayGainUpdater(logging)
print rgu.main()
except Exception, e:
print e
print traceback.format_exc()