# -*- Mode: Python; test-case-name: morituri.test.test_image_image -*- # vi:si:et:sw=4:sts=4:ts=4 # Morituri - for those about to RIP # Copyright (C) 2009 Thomas Vander Stichele # This file is part of morituri. # # morituri 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 3 of the License, or # (at your option) any later version. # # morituri 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 morituri. If not, see . """ Wrap on-disk CD images based on the .cue file. """ import os import struct import gst from morituri.common import task, checksum from morituri.image import cue, toc class Image: def __init__(self, path): """ @param path: .cue path """ self._path = path self.cue = cue.Cue(path) self.cue.parse() self._offsets = [] # 0 .. trackCount - 1 self._lengths = [] # 0 .. trackCount - 1 def getRealPath(self, path): """ Translate the .cue's FILE to an existing path. """ if os.path.exists(path): return path # .cue FILE statements have Windows-style path separators, so convert tpath = os.path.join(*path.split('\\')) candidatePaths = [] # if the path is relative: # - check relatively to the cue file # - check only the filename part relative to the cue file if tpath == os.path.abspath(tpath): candidatePaths.append(tPath) else: candidatePaths.append(os.path.join( os.path.dirname(self._path), tpath)) candidatePaths.append(os.path.join( os.path.dirname(self._path), os.path.basename(tpath))) for candidate in candidatePaths: noext, _ = os.path.splitext(candidate) for ext in ['wav', 'flac']: cpath = '%s.%s' % (noext, ext) if os.path.exists(cpath): return cpath raise KeyError, "Cannot find file for %s" % path def setup(self, runner): """ Do initial setup, like figuring out track lengths. """ verify = ImageVerifyTask(self) runner.run(verify) # calculate offset and length for each track # CD's have a standard lead-in time of 2 seconds; # checksums that use it should add it there offset = self.cue.tracks[0].getIndex(1)[0] tracks = [] for i in range(len(self.cue.tracks)): length = self.cue.getTrackLength(self.cue.tracks[i]) if length == -1: length = verify.lengths[i + 1] tracks.append(toc.Track(i + 1, offset, offset + length - 1)) offset += length self.toc = toc.TOC(tracks) class AccurateRipChecksumTask(task.MultiTask): """ I calculate the AccurateRip checksums of all tracks. """ description = "Checksumming tracks" def __init__(self, image): self._image = image cue = image.cue self.checksums = [] for trackIndex, track in enumerate(cue.tracks): index = track._indexes[1] length = cue.getTrackLength(track) file = index[1] offset = index[0] path = image.getRealPath(file.path) checksumTask = checksum.AccurateRipChecksumTask(path, trackNumber=trackIndex + 1, trackCount=len(cue.tracks), frameStart=offset * checksum.SAMPLES_PER_FRAME, frameLength=length * checksum.SAMPLES_PER_FRAME) self.addTask(checksumTask) def stop(self): self.checksums = [t.checksum for t in self.tasks] task.MultiTask.stop(self) class AudioLengthTask(task.Task): """ I calculate the length of a track in audio frames. @ivar length: length of the decoded audio file, in audio frames. """ length = None def __init__(self, path): self._path = path def debug(self, *args, **kwargs): return print args, kwargs def start(self, runner): task.Task.start(self, runner) self._pipeline = gst.parse_launch(''' filesrc location="%s" ! decodebin ! audio/x-raw-int ! fakesink name=sink''' % self._path) self.debug('pausing') self._pipeline.set_state(gst.STATE_PAUSED) self._pipeline.get_state() self.debug('paused') self.debug('query duration') sink = self._pipeline.get_by_name('sink') assert sink, 'Error constructing pipeline' length, format = sink.query_duration(gst.FORMAT_DEFAULT) # wavparse 0.10.14 returns in bytes if format == gst.FORMAT_BYTES: self.debug('query returned in BYTES format') length /= 4 self.debug('total length', length) self.length = length self._pipeline.set_state(gst.STATE_NULL) self.stop() class ImageVerifyTask(task.MultiTask): """ I verify a disk image and get the necessary track lengths. """ description = "Checking tracks" lengths = None def __init__(self, image): self._image = image cue = image.cue self._tasks = [] self.lengths = {} for trackIndex, track in enumerate(cue.tracks): index = track._indexes[1] offset = index[0] length = cue.getTrackLength(track) file = index[1] if length == -1: path = image.getRealPath(file.path) taskk = AudioLengthTask(path) self.addTask(taskk) self._tasks.append((trackIndex + 1, track, taskk)) def stop(self): for trackIndex, track, taskk in self._tasks: # print '%d has length %d' % (trackIndex, taskk.length) index = track._indexes[1] offset = index[0] assert taskk.length % checksum.SAMPLES_PER_FRAME == 0 end = taskk.length / checksum.SAMPLES_PER_FRAME self.lengths[trackIndex] = end - offset task.MultiTask.stop(self) # FIXME: move this method to a different module ? def getAccurateRipResponses(data): ret = [] while data: trackCount = struct.unpack("B", data[0])[0] bytes = 1 + 12 + trackCount * (1 + 8) ret.append(AccurateRipResponse(data[:bytes])) data = data[bytes:] return ret class AccurateRipResponse(object): """ I represent the response of the AccurateRip online database. """ trackCount = None discId1 = "" discId2 = "" cddbDiscId = "" confidences = None checksums = None def __init__(self, data): self.trackCount = struct.unpack("B", data[0])[0] self.discId1 = "%08x" % struct.unpack("