trackcmp

#!/usr/bin/python

#Audio Tools, a module and set of tools for manipulating audio data
#Copyright (C) 2007-2012  Brian Langenberger

#This program is free software; you can redistribute it and/or modify
#it under the terms of the GNU General Public License as published by
#the Free Software Foundation; either version 2 of the License, or
#(at your option) any later version.

#This program is distributed in the hope that it will be useful,
#but WITHOUT ANY WARRANTY; without even the implied warranty of
#MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#GNU General Public License for more details.

#You should have received a copy of the GNU General Public License
#along with this program; if not, write to the Free Software
#Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA


import audiotools
import sys
import os
import os.path
import operator
import gettext

gettext.install("audiotools", unicode=True)


def near_equal(audiofile1, audiofile2):
    p1 = audiofile1.to_pcm()
    p2 = audiofile2.to_pcm()
    try:
        return audiotools.stripped_pcm_cmp(p1, p2)
    finally:
        p1.close()
        p2.close()


def cmp_files(progress, audiofile1, audiofile2):
    """Returns (path1, path2, mismatch) tuple

    where mismatch is the int of the first PCM mismatch,
    None if the files match exactly or
    a negative value if some error occurs."""

    try:
        if (os.path.samefile(audiofile1.filename, audiofile2.filename)):
            return (audiofile1.filename,
                    audiofile2.filename,
                    None)
        else:
            return (audiofile1.filename,
                    audiofile2.filename,
                    audiotools.pcm_frame_cmp(
                    audiotools.to_pcm_progress(audiofile1,
                                               progress),
                    audiofile2.to_pcm()))
    except (IOError, ValueError, audiotools.DecodingError):
        return (audiofile1.filename,
                audiofile2.filename,
                -1)


class Results:
    def __init__(self, messenger):
        self.successes = 0
        self.failures = 0
        self.msg = messenger

    def missing(self, directory, track_number, album_number):
        self.failures += 1
        if (album_number is None):
            self.msg.info(_(u"%(path)s : %(result)s") % {
                    "path":audiotools.Filename(
                        os.path.join(directory,
                                     "track %2.2d" % (
                                track_number
                                if track_number is not None else 0))),
                    "result": self.msg.ansi(_(u"missing"), [self.msg.FG_RED])})
        else:
            self.msg.info(_(u"%(path)s : %(result)s") % {
                    "path":audiotools.Filename(
                        os.path.join(directory,
                                     "album %d track %2.2d" % (
                                album_number,
                                track_number
                                if track_number is not None else 0))),
                    "result": self.msg.ansi(_(u"missing"), [self.msg.FG_RED])})
        sys.stdout.flush()

    def cmp_result(self, result):
        (path1, path2, mismatch) = result

        if (mismatch is None):
            self.successes += 1

            return u"%(path1)s <> %(path2)s : %(result)s" % {
                "path1": audiotools.Filename(path1),
                "path2": audiotools.Filename(path2),
                "result": self.msg.ansi(
                    _(u"OK"),
                    [self.msg.FG_GREEN])}
        elif (mismatch >= 0):
            self.failures += 1

            return u"%(path1)s <> %(path2)s : %(result)s" % {
                "path1": audiotools.Filename(path1),
                "path2": audiotools.Filename(path2),
                "result": self.msg.ansi(
                    _(u"differ at PCM frame %d") % (mismatch + 1),
                    [self.msg.FG_RED])}
        else:
            self.failures += 1

            return u"%(path1)s <> %(path2)s : %(result)s" % {
                "path1": audiotools.Filename(path1),
                "path2": audiotools.Filename(path2),
                "result": self.msg.ansi(
                    _(u"error"),
                    [self.msg.FG_RED])}

if (__name__ == '__main__'):
    parser = audiotools.OptionParser(
        usage=_(u'%prog <file 1> <file 2>'),
        version="Python Audio Tools %s" % (audiotools.VERSION))

    parser.add_option(
        '-j', '--joint',
        action='store',
        type='int',
        default=audiotools.MAX_JOBS,
        dest='max_processes',
        help=_(u'the maximum number of processes to run at a time'))

    parser.add_option('-R', '--no-summary',
                      action='store_true',
                      dest='no_summary',
                      help=_(u'suppress summary output'))

    parser.add_option('-V', '--verbose',
                      action='store',
                      dest='verbosity',
                      choices=audiotools.VERBOSITY_LEVELS,
                      default=audiotools.DEFAULT_VERBOSITY,
                      help=_(u'the verbosity level to execute at'))

    (options, args) = parser.parse_args()
    msg = audiotools.Messenger("trackcmp", options)

    if (options.max_processes < 1):
        msg.error(_(u'You must run at least 1 process at a time'))
        sys.exit(1)

    check_function = audiotools.pcm_frame_cmp

    if (len(args) == 2):
        if (os.path.isfile(args[0]) and os.path.isfile(args[1])):
            #comparing two files

            audiofiles = audiotools.open_files(args,
                                               messenger=msg,
                                               sorted=False)
            if (len(audiofiles) != 2):
                msg.error(_(u"Both files to be compared must be audio files"))
                sys.exit(1)
            try:
                (audiofile_1,
                 audiofile_2) = audiofiles
                filename_1 = audiotools.Filename(args[0])
                filename_2 = audiotools.Filename(args[1])

                if (filename_1 == filename_2):
                    #both files are the same so they must be identical
                    pass
                else:
                    frame_mismatch = check_function(audiofile_1.to_pcm(),
                                                    audiofile_2.to_pcm())
                    if (frame_mismatch is None):
                        #all PCM frames the same
                        #so files are identical
                        pass
                    else:
                        msg.partial_info(
                            _(u"%(file1)s <> %(file2)s : ") % \
                                {"file1": filename_1,
                                 "file2": filename_2})
                        msg.info(msg.ansi(
                                _(u"differ at PCM frame %(frame_number)d") %
                                {"frame_number": frame_mismatch + 1},
                                [msg.FG_RED]))
                        sys.exit(1)
            except (IOError, ValueError, audiotools.DecodingError), err:
                msg.error(unicode(err))
                sys.exit(1)

        elif (os.path.isdir(args[0]) and os.path.isdir(args[1])):
            #comparing two directories
            results = Results(msg)

            files1 = audiotools.open_files(
                [os.path.join(args[0], f) for f in os.listdir(args[0])
                 if os.path.isfile(os.path.join(args[0], f))],
                messenger=msg)
            files2 = audiotools.open_files(
                [os.path.join(args[1], f) for f in os.listdir(args[1])
                 if os.path.isfile(os.path.join(args[1], f))],
                messenger=msg)

            files1_map = dict([((f.album_number(), f.track_number()), f)
                               for f in files1])
            files2_map = dict([((f.album_number(), f.track_number()), f)
                               for f in files2])

            files1_tracknumbers = set(files1_map.keys())
            files2_tracknumbers = set(files2_map.keys())

            queue = audiotools.ExecProgressQueue(
                audiotools.ProgressDisplay(msg))

            for (album_number, track_number) in sorted(
                list(files1_tracknumbers - files2_tracknumbers)):
                results.missing(args[1], track_number, album_number)

            for (album_number, track_number) in sorted(
                list(files2_tracknumbers - files1_tracknumbers)):
                results.missing(args[0], track_number, album_number)

            for (album_number, track_number) in sorted(
                list(files1_tracknumbers & files2_tracknumbers)):
                queue.execute(
                    function=cmp_files,
                    progress_text=(u"%s <> %s" %
                                   (audiotools.Filename(
                                files1_map[(album_number,
                                            track_number)].filename),
                                    audiotools.Filename(
                                files2_map[(album_number,
                                            track_number)].filename))),
                    completion_output=results.cmp_result,
                    audiofile1=files1_map[(album_number, track_number)],
                    audiofile2=files2_map[(album_number, track_number)])

            queue.run(options.max_processes)

            if (not options.no_summary):
                msg.output(_(u"Results:"))
                msg.output(u"")
                msg.new_row()
                msg.output_column(_(u"success"), True)
                msg.output_column(u" ")
                msg.output_column(_(u"failure"), True)
                msg.output_column(u" ")
                msg.output_column(_(u"total"), True)
                msg.divider_row([u"-", u" ", u"-", u" ", u"-"])
                msg.new_row()
                msg.output_column(unicode(results.successes), True)
                msg.output_column(u" ")
                msg.output_column(unicode(results.failures), True)
                msg.output_column(u" ")
                msg.output_column(unicode(results.successes +
                                          results.failures), True)
                msg.output_rows()

            if (results.failures > 0):
                sys.exit(1)
        else:
            #comparison mismatch
            msg.error(_(u"%(file1)s <> %(file2)s : %(result)s") %
                      {"file1": audiotools.Filename(args[0]),
                       "file2": audiotools.Filename(args[1]),
                       "result": msg.ansi(
                        _(u"must be either files or directories"),
                        [msg.FG_RED])})
            sys.exit(1)
    elif (len(args) > 2):
        #possibly comparing disk image against tracks
        progress = audiotools.SingleProgressDisplay(msg, u"")
        progress.delete_row(0)

        audiofiles = audiotools.open_files(args, messenger=msg, sorted=False)

        audiofiles.sort(lambda t1, t2: cmp(t1.total_frames(),
                                           t2.total_frames()))

        if (sum([t.total_frames() for t in audiofiles[0:-1]]) !=
            audiofiles[-1].total_frames()):
            msg.usage(u"<CD image> <track 1> <track 2> ...")
            sys.exit(1)

        cd_image = audiofiles[-1]
        tracks = audiofiles[0:-1]

        #all tracks should have the same album number and track total
        tracks.sort(lambda t1, t2: cmp(t1.track_number(),
                                       t2.track_number()))

        cd_data = audiotools.BufferedPCMReader(cd_image.to_pcm())
        for (i, track) in enumerate(tracks):
            progress.add_row(0,
                             u"%(file1)s <> %(file2)s" %
                             {"file1": audiotools.Filename(cd_image.filename),
                              "file2": audiotools.Filename(track.filename)})
            mismatch = audiotools.pcm_frame_cmp(
                audiotools.to_pcm_progress(track, progress.update),
                audiotools.LimitedPCMReader(cd_data, track.total_frames()))
            progress.delete_row(0)
            progress.clear()
            if (mismatch is None):
                msg.output(
                    audiotools.output_progress(
                        u"%(path1)s <> %(path2)s : %(result)s" % {
                            "path1": audiotools.Filename(cd_image.filename),
                            "path2": audiotools.Filename(track.filename),
                            "result": msg.ansi(
                                _(u"OK"),
                                [msg.FG_GREEN])},
                        i + 1, len(tracks)))
            else:
                msg.output(
                    audiotools.output_progress(
                        u"%(path1)s <> %(path2)s : %(result)s" % {
                            "path1": audiotools.Filename(cd_image.filename),
                            "path2": audiotools.Filename(track.filename),
                            "result": msg.ansi(
                                _(u"differ at PCM frame %d") % (mismatch + 1),
                                [msg.FG_RED])},
                        i + 1, len(tracks)))
                sys.exit(1)
    else:
        msg.usage(u"<track 1> <track 2>")
        msg.exit(1)