Sox (#48)

* update README to force python2
* add support for PRE_EMPHASIS detection; mark pre-emphasis as FLAGS PRE in .cue
* correct typo, s/ivar/type/ for pre_emphasis element in track table
* add SoxPeakTask, program/sox module, sox.peak_level()
* use sox peak detection in lieu of gstreamer

morituri/common/encode.py

@@ -30,6 +30,7 @@
 from morituri.common import task as ctask
 
 from morituri.extern.task import task, gstreamer
+from morituri.program import sox
 
 
 class Profile(log.Loggable):
@@ -164,6 +165,20 @@ class VorbisProfile(Profile):
 ALL_PROFILES = PROFILES.copy()
 ALL_PROFILES.update(LOSSY_PROFILES)
 
+class SoxPeakTask(task.Task):
+ description = 'Calculating peak level'
+
+ def __init__(self, track_path):
+ self.track_path = track_path
+ self.peak = None
+
+ def start(self, runner):
+ task.Task.start(self, runner)
+ self.schedule(0.0, self._sox_peak)
+
+ def _sox_peak(self):
+ self.peak = sox.peak_level(self.track_path)
+ self.stop()
 
 class EncodeTask(ctask.GstPipelineTask):
  """

morituri/program/cdparanoia.py

@@ -492,6 +492,7 @@ def __init__(self, path, table, start, stop, overread, offset=0,
  taglist=taglist, what=what))
  # make sure our encoding is accurate
  self.tasks.append(checksum.CRC32Task(tmpoutpath))
+ self.tasks.append(encode.SoxPeakTask(tmppath))
 
  self.checksum = None
 
@@ -502,7 +503,7 @@ def stop(self):
  if not self.exception:
  self.quality = max(self.tasks[0].quality,
  self.tasks[2].quality)
- self.peak = self.tasks[4].peak
+ self.peak = self.tasks[6].peak
  self.debug('peak: %r', self.peak)
  self.testspeed = self.tasks[0].speed
  self.copyspeed = self.tasks[2].speed

morituri/program/sox.py

@@ -0,0 +1,16 @@
+import os
+import logging
+from subprocess import Popen, PIPE
+
+SOX = 'sox'
+
+def peak_level(track_path):
+ if not os.path.exists(track_path):
+ logging.warning("SoX peak detection failed: file not found")
+ return None
+ sox = Popen([SOX, track_path, "-n", "stat"], stderr=PIPE)
+ out, err = sox.communicate()
+ if sox.returncode:
+ logging.warning("SoX peak detection failed: " + s.returncode)
+ return None
+ return float(err.split('\n')[3].split()[2]) # Maximum amplitude: 0.123456

morituri/rip/cd.py

@@ -430,7 +430,7 @@ def ripIfNotRipped(number):
  number)
  raise
 
- self.stdout.write('Peak level: {:.2%} \n'.format(math.sqrt(trackResult.peak)))
+ self.stdout.write('Peak level: {:.2%} \n'.format(trackResult.peak))
 
  self.stdout.write('Rip quality: {:.2%}\n'.format(trackResult.quality))