trackcmp

#!/usr/bin/python

#Audio Tools, a module and set of tools for manipulating audio data
#Copyright (C) 2007-2013  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 sys
import os
import os.path
import operator
import audiotools
import audiotools.text as _


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)


def cmp_result(result):
    msg = audiotools.Messenger("trackcmp", None)
    ansi = msg.ansi
    FG_GREEN = msg.FG_GREEN
    FG_RED = msg.FG_RED

    (path1, path2, mismatch) = result

    if (mismatch is None):
        return ((_.LAB_TRACKCMP_CMP %
                 {"file1": audiotools.Filename(path1),
                  "file2": audiotools.Filename(path2)}) +
                u" : " + ansi(_.LAB_TRACKCMP_OK, [FG_GREEN]))
    elif (mismatch >= 0):
        return ((_.LAB_TRACKCMP_CMP %
                 {"file1": audiotools.Filename(path1),
                  "file2": audiotools.Filename(path2)}) +
                u" : " + ansi(_.LAB_TRACKCMP_MISMATCH %
                              {"frame_number": mismatch + 1},
                              [FG_RED]))
    else:
        return ((_.LAB_TRACKCMP_CMP %
                 {"file1": audiotools.Filename(path1),
                  "file2": audiotools.Filename(path2)}) +
                u" : " + ansi(_.LAB_TRACKCMP_ERROR, [FG_RED]))


def image_compare(progress, image_audiofile, track_audiofile,
                  image_filename, track_filename,
                  pcm_frames_offset, total_pcm_frames):
    image_pcmreader = image_audiofile.to_pcm()

    #if image_pcmreader has seek(),
    #use it to reduce the amount of frames to skip
    if (hasattr(image_pcmreader, "seek") and callable(image_pcmreader.seek)):
        pcm_frames_offset -= image_pcmreader.seek(pcm_frames_offset)

    return (
        audiotools.pcm_frame_cmp(
            audiotools.PCMReaderWindow(image_pcmreader,
                                       pcm_frames_offset,
                                       total_pcm_frames),
            audiotools.PCMReaderProgress(track_audiofile.to_pcm(),
                                         total_pcm_frames,
                                         progress)),
        unicode(image_filename),
        unicode(track_filename))


def image_compare_results(result):
    msg = audiotools.Messenger("trackcmp", None)
    ansi = msg.ansi
    FG_GREEN = msg.FG_GREEN
    FG_RED = msg.FG_RED

    (mismatch, image_name, track_name) = result
    return u"%s : %s" % (_.LAB_TRACKCMP_CMP %
                         {"file1": image_name, "file2": track_name},
                         (ansi(_.LAB_TRACKCMP_OK, [FG_GREEN]) if
                          mismatch is None else
                          ansi(_.LAB_TRACKCMP_MISMATCH %
                               {"frame_number": mismatch + 1}, [FG_RED])))


if (__name__ == '__main__'):
    import argparse

    parser = argparse.ArgumentParser(description=_.DESCRIPTION_TRACKCMP)

    parser.add_argument("--version",
                        action="version",
                        version="Python Audio Tools %s" % (audiotools.VERSION))

    parser.add_argument("-V", "--verbose",
                        action="store",
                        dest="verbosity",
                        choices=audiotools.VERBOSITY_LEVELS,
                        default=audiotools.DEFAULT_VERBOSITY,
                        help=_.OPT_VERBOSE)

    parser.add_argument("-j", "--joint",
                        type=int,
                        default=audiotools.MAX_JOBS,
                        dest="max_processes",
                        help=_.OPT_JOINT)

    parser.add_argument("-S", "--no-summary",
                        action="store_true",
                        dest="no_summary",
                        help=_.OPT_NO_SUMMARY)

    parser.add_argument("filename",
                        metavar="PATH",
                        help=_.OPT_INPUT_FILENAME_OR_IMAGE)

    parser.add_argument("filenames",
                        metavar="PATH",
                        nargs="+",
                        help=_.OPT_INPUT_FILENAME_OR_DIR)

    options = parser.parse_args()

    msg = audiotools.Messenger("trackcmp", options)

    args = [options.filename] + options.filenames

    if (options.max_processes < 1):
        msg.error(_.ERR_INVALID_JOINT)
        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(_.ERR_TRACKCMP_TYPE_MISMATCH)
                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(
                            (_.LAB_TRACKCMP_CMP %
                             {"file1": filename_1,
                              "file2": filename_2}) +
                            u" : ")
                        msg.info(
                            msg.ansi(
                                _.LAB_TRACKCMP_MISMATCH %
                                {"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

            to_compare = []
            results = []

            files1 = dict(
                [(f.filename, f) for f in
                 audiotools.open_files(
                     [path for path in [os.path.join(args[0], f) for f in
                                        os.listdir(args[0])] if
                      os.path.isfile(path)],
                     sorted=False,
                     messenger=msg)])

            files2 = dict(
                [(f.filename, f) for f in
                 audiotools.open_files(
                     [path for path in [os.path.join(args[1], f) for f in
                                        os.listdir(args[1])] if
                      os.path.isfile(path)],
                     sorted=False,
                     messenger=msg)])

            #first, attempt to match files by their stream characteristics
            streams1 = {}
            streams2 = {}

            for (files, streams) in [(files1, streams1),
                                     (files2, streams2)]:
                for f in files.values():
                    streams.setdefault((f.bits_per_sample(),
                                        f.channels(),
                                        f.sample_rate(),
                                        f.total_frames()), []).append(f)

            #anything with matching specs
            #and only a single possible match per directory
            #is queued for comparison
            for specs in set(streams1.keys()) & set(streams2.keys()):
                if (((len(streams1[specs]) == 1) and
                     (len(streams2[specs]) == 1))):
                    file1 = streams1[specs][0]
                    file2 = streams2[specs][0]

                    #remove matched files from lists
                    del(files1[file1.filename])
                    del(files2[file2.filename])

                    #queue up comparison job
                    to_compare.append((file1, file2))

            #then, attempt to match leftover files by metadata
            #such as album_number and track_number
            metadatas1 = {}
            metadatas2 = {}

            for (files, metadatas) in [(files1, metadatas1),
                                       (files2, metadatas2)]:
                for f in files.values():
                    m = f.get_metadata()
                    if (m is not None):
                        metadatas.setdefault((m.track_number,
                                              m.album_number), []).append(f)
                    else:
                        metadatas.setdefault((None,
                                              None), []).append(f)

            for metadata in set(metadatas1.keys()) & set(metadatas2.keys()):
                if (((len(metadatas1[metadata]) == 1) and
                     (len(metadatas2[metadata]) == 1))):
                    file1 = metadatas1[metadata][0]
                    file2 = metadatas2[metadata][0]

                    #remove matched files from lists
                    del(files1[file1.filename])
                    del(files2[file2.filename])

                    #queue up comparison job
                    to_compare.append((file1, file2))

            #anything left over is marked as a missing file
            for (files, directory) in [(files1, args[1]), (files2, args[0])]:
                for filename in files.keys():
                    msg.info(
                        msg.ansi(
                            _.LAB_TRACKCMP_MISSING %
                            {"filename": audiotools.Filename(
                                    os.path.basename(filename)),
                             "directory": audiotools.Filename(
                                    directory)},
                            [msg.FG_RED]))
                    sys.stdout.flush()
                    results.append((filename, None, 0))

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

            to_compare.sort(lambda f1, f2: cmp(f1[0].filename, f2[0].filename))

            for (track1, track2) in to_compare:
                queue.execute(
                    function=cmp_files,
                    progress_text=_.LAB_TRACKCMP_CMP %
                    {"file1": audiotools.Filename(track1.filename),
                     "file2": audiotools.Filename(track2.filename)},
                    completion_output=cmp_result,
                    audiofile1=track1,
                    audiofile2=track2)

            results.extend(queue.run(options.max_processes))
            successes = len([r for r in results if r[2] is None])
            failures = len(results) - successes

            if (not options.no_summary):
                msg.output(_.LAB_TRACKCMP_RESULTS)
                msg.output(u"")
                msg.new_row()
                msg.output_column(_.LAB_TRACKCMP_HEADER_SUCCESS, True)
                msg.output_column(u" ")
                msg.output_column(_.LAB_TRACKCMP_HEADER_FAILURE, True)
                msg.output_column(u" ")
                msg.output_column(_.LAB_TRACKCMP_HEADER_TOTAL, True)
                msg.divider_row([u"-", u" ", u"-", u" ", u"-"])
                msg.new_row()
                msg.output_column(unicode(successes), True)
                msg.output_column(u" ")
                msg.output_column(unicode(failures), True)
                msg.output_column(u" ")
                msg.output_column(unicode(successes + failures), True)
                msg.output_rows()

            if (failures > 0):
                sys.exit(1)
        else:
            #comparison mismatch
            msg.error((_.LAB_TRACKCMP_CMP %
                       {"file1": audiotools.Filename(args[0]),
                        "file2": audiotools.Filename(args[1])}) +
                      u" : " +
                      msg.ansi(_.LAB_TRACKCMP_TYPE_MISMATCH,
                               [msg.FG_RED]))
            sys.exit(1)
    elif (len(args) > 2):
        #possibly comparing disk image against tracks
        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(_.USAGE_TRACKCMP_CDIMAGE)
            sys.exit(1)

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

        image_name = audiotools.Filename(cd_image.filename)

        #all tracks should have the same album number and track total
        tracks = audiotools.sorted_tracks(tracks)

        cd_data = audiotools.BufferedPCMReader(cd_image.to_pcm())

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

        pcm_offset = 0

        for (i, track) in enumerate(tracks):
            track_name = audiotools.Filename(track.filename)
            queue.execute(
                function=image_compare,
                progress_text=_.LAB_TRACKCMP_CMP %
                {"file1": image_name, "file2": track_name},
                completion_output=image_compare_results,
                image_audiofile=cd_image,
                track_audiofile=track,
                image_filename=image_name,
                track_filename=track_name,
                pcm_frames_offset=pcm_offset,
                total_pcm_frames=track.total_frames())

            pcm_offset += track.total_frames()

        if (set([r[0] for r in
                 queue.run(options.max_processes)]) != set([None])):
            sys.exit(1)
    else:
        msg.usage(_.USAGE_TRACKCMP_FILES)
        sys.exit(1)