benya/whipper-gui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

* examples/ARcue.py:

Browse Source 
* morituri/common/checksum.py:
	* morituri/image/image.py:
	* morituri/test/test_image_image.py:
	  Change crc to checksum everywhere.
This commit is contained in:
Thomas Vander Stichele
2009-04-15 13:31:35 +00:00
parent 9b5050d221
commit b63ef021a7
5 changed files with 83 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ChangeLog
View File

@@ -1,3 +1,11 @@
2009-04-15 Thomas Vander Stichele <thomas at apestaart dot org>
* examples/ARcue.py:
* morituri/common/checksum.py:
* morituri/image/image.py:
* morituri/test/test_image_image.py:
Change crc to checksum everywhere.
2009-04-15 Thomas Vander Stichele <thomas at apestaart dot org>
* morituri/image/image.py:

20
examples/ARcue.py
View File

@@ -29,7 +29,7 @@ gobject.threads_init()
import gtk
from morituri.image import image
from morituri.common import task, crc
from morituri.common import task, checksum
def gtkmain(runner, taskk):
runner.connect('stop', lambda _: gtk.main_quit())
@@ -66,7 +66,7 @@ def main(argv):
cueImage = image.Image(path)
verifytask = image.ImageVerifyTask(cueImage)
cuetask = image.AudioRipCRCTask(cueImage)
cuetask = image.AccurateRipChecksumTask(cueImage)
if options.runner == 'cli':
runner = task.SyncRunner()
@@ -105,22 +105,22 @@ def main(argv):
response = None
for i, crc in enumerate(cuetask.crcs):
for i, checksum in enumerate(cuetask.checksums):
status = 'rip NOT accurate'
confidence = None
arcrc = None
archecksum = None
for j, r in enumerate(responses):
if "%08x" % crc == r.crcs[i]:
if "%08x" % checksum == r.checksums[i]:
if not response:
response = r
else:
assert r == response, \
"CRC %s for %d matches wrong response %d, crc %s" % (
crc, i + 1, j + 1, response.crcs[i])
"CRC %s for %d matches wrong response %d, checksum %s" % (
checksum, i + 1, j + 1, response.checksums[i])
status = 'rip accurate '
arcrc = crc
archecksum = checksum
confidence = response.confidences[i]
c = "(not found)"
@@ -133,9 +133,9 @@ def main(argv):
if confidence < maxConfidence:
c = "(confidence %3d of %3d)" % (confidence, maxConfidence)
ar = " AR [%s]" % response.crcs[i]
ar = " AR [%s]" % response.checksums[i]
print "Track %2d: %s %s mine [%08x] %s" % (
i + 1, status, c, crc, ar)
i + 1, status, c, checksum, ar)
main(sys.argv)

61
morituri/common/checksum.py
View File

@@ -1,4 +1,4 @@
# -*- Mode: Python; test-case-name: morituri.test.test_common_crc -*-
# -*- Mode: Python; test-case-name: morituri.test.test_common_checksum -*-
# vi:si:et:sw=4:sts=4:ts=4
# Morituri - for those about to RIP
@@ -34,9 +34,15 @@ FRAMES_PER_DISC_FRAME = 588
SAMPLES_PER_DISC_FRAME = FRAMES_PER_DISC_FRAME * 4
DISC_FRAMES_PER_SECOND = 75
class CRCTask(task.Task):
class ChecksumTask(task.Task):
"""
I am a task that calculates a checksum.
@ivar checksum: the resulting checksum
"""
# this object needs a main loop to stop
description = 'Calculating CRC checksum...'
description = 'Calculating checksum...'
def __init__(self, path, frameStart=0, frameLength=-1):
"""
@@ -54,13 +60,13 @@ class CRCTask(task.Task):
self._frameStart = frameStart
self._frameLength = frameLength
self._frameEnd = None
self._crc = 0
self._checksum = 0
self._bytes = 0 # number of bytes received
self._first = None
self._last = None
self._adapter = gst.Adapter()
self.crc = None # result
self.checksum = None # result
def start(self, runner):
task.Task.start(self, runner)
@@ -139,7 +145,7 @@ class CRCTask(task.Task):
# FIXME: in 0.10.14.1, take_buffer leaks a ref
buffer = self._adapter.take_buffer(SAMPLES_PER_DISC_FRAME)
self._crc = self.do_crc_buffer(buffer, self._crc)
self._checksum = self.do_checksum_buffer(buffer, self._checksum)
self._bytes += len(buffer)
# update progress
@@ -149,7 +155,7 @@ class CRCTask(task.Task):
# marshall to the main thread
self.runner.schedule(0, self.setProgress, progress)
def do_crc_buffer(self, buffer, crc):
def do_checksum_buffer(self, buffer, checksum):
"""
Subclasses should implement this.
"""
@@ -171,35 +177,40 @@ class CRCTask(task.Task):
print 'ERROR: not a single buffer gotten'
raise
else:
self._crc = self._crc % 2 ** 32
self._checksum = self._checksum % 2 ** 32
last = self._last.offset + len(self._last) / 4 - 1
self.debug("last sample:", last)
self.debug("frame length:", self._frameLength)
self.debug("CRC: %08X" % self._crc)
self.debug("checksum: %08X" % self._checksum)
self.debug("bytes: %d" % self._bytes)
if self._frameEnd != last:
print 'ERROR: did not get all frames, %d missing' % (
self._frameEnd - last)
# publicize and stop
self.crc = self._crc
self.checksum = self._checksum
task.Task.stop(self)
class CRC32Task(CRCTask):
class CRC32Task(ChecksumTask):
"""
I do a simple CRC32 check.
"""
def do_crc_buffer(self, buffer, crc):
return zlib.crc32(buffer, crc)
def do_checksum_buffer(self, buffer, checksum):
return zlib.crc32(buffer, checksum)
class CRCAudioRipTask(CRCTask):
class AccurateRipChecksumTask(ChecksumTask):
"""
I implement the AccurateRip checksum.
See http://www.accuraterip.com/
"""
def __init__(self, path, trackNumber, trackCount, frameStart=0, frameLength=-1):
CRCTask.__init__(self, path, frameStart, frameLength)
ChecksumTask.__init__(self, path, frameStart, frameLength)
self._trackNumber = trackNumber
self._trackCount = trackCount
self._discFrameCounter = 0 # 1-based
def do_crc_buffer(self, buffer, crc):
def do_checksum_buffer(self, buffer, checksum):
self._discFrameCounter += 1
# on first track ...
@@ -207,25 +218,25 @@ class CRCAudioRipTask(CRCTask):
# ... skip first 4 CD frames
if self._discFrameCounter <= 4:
gst.debug('skipping frame %d' % self._discFrameCounter)
return crc
return checksum
# ... on 5th frame, only use last value
elif self._discFrameCounter == 5:
values = struct.unpack("<I", buffer[-4:])
crc += FRAMES_PER_DISC_FRAME * 5 * values[0]
crc &= 0xFFFFFFFF
return crc
checksum += FRAMES_PER_DISC_FRAME * 5 * values[0]
checksum &= 0xFFFFFFFF
return checksum
# on last track, skip last 5 CD frames
if self._trackNumber == self._trackCount:
discFrameLength = self._frameLength / FRAMES_PER_DISC_FRAME
if self._discFrameCounter > discFrameLength - 5:
self.debug('skipping frame %d' % self._discFrameCounter)
return crc
return checksum
values = struct.unpack("<%dI" % (len(buffer) / 4), buffer)
sum = 0
for i, value in enumerate(values):
# self._bytes is updated after do_crc_buffer
# self._bytes is updated after do_checksum_buffer
sum += (self._bytes / 4 + i + 1) * value
sum &= 0xFFFFFFFF
# offset = self._bytes / 4 + i + 1
@@ -233,6 +244,6 @@ class CRCAudioRipTask(CRCTask):
# print 'THOMAS: frame %d, ends before %d, last value %08x, CRC %08x' % (
# offset / FRAMES_PER_DISC_FRAME, offset, value, sum)
crc += sum
crc &= 0xFFFFFFFF
return crc
checksum += sum
checksum &= 0xFFFFFFFF
return checksum

22
morituri/image/image.py
View File

@@ -150,17 +150,17 @@ class MultiTask(task.Task):
self._next()
class AudioRipCRCTask(MultiTask):
class AccurateRipChecksumTask(MultiTask):
"""
I calculate the AudioRip CRC's of all tracks.
I calculate the AccurateRip checksums of all tracks.
"""
description = "CRC'ing tracks"
description = "Checksumming tracks"
def __init__(self, image):
self._image = image
cue = image.cue
self.crcs = []
self.checksums = []
for trackIndex, track in enumerate(cue.tracks):
index = track._indexes[1]
@@ -169,14 +169,14 @@ class AudioRipCRCTask(MultiTask):
offset = index[0]
path = image.getRealPath(file.path)
crctask = checksum.CRCAudioRipTask(path,
checksumTask = checksum.AccurateRipChecksumTask(path,
trackNumber=trackIndex + 1, trackCount=len(cue.tracks),
frameStart=offset * checksum.FRAMES_PER_DISC_FRAME,
frameLength=length * checksum.FRAMES_PER_DISC_FRAME)
self.addTask(crctask)
self.addTask(checksumTask)
def stop(self):
self.crcs = [t.crc for t in self.tasks]
self.checksums = [t.checksum for t in self.tasks]
MultiTask.stop(self)
class AudioLengthTask(task.Task):
@@ -281,7 +281,7 @@ class AccurateRipResponse(object):
discId2 = ""
cddbDiscId = ""
confidences = None
crcs = None
checksums = None
def __init__(self, data):
self.trackCount = struct.unpack("B", data[0])[0]
@@ -290,12 +290,12 @@ class AccurateRipResponse(object):
self.cddbDiscId = "%08x" % struct.unpack("<L", data[9:13])[0]
self.confidences = []
self.crcs = []
self.checksums = []
pos = 13
for i in range(self.trackCount):
confidence = struct.unpack("B", data[pos])[0]
crc = "%08x" % struct.unpack("<L", data[pos + 1:pos + 5])[0]
checksum = "%08x" % struct.unpack("<L", data[pos + 1:pos + 5])[0]
pos += 9
self.confidences.append(confidence)
self.crcs.append(crc)
self.checksums.append(checksum)

36
morituri/test/test_image_image.py
View File

@@ -20,15 +20,15 @@ class TrackSingleTestCase(unittest.TestCase):
self.runner = task.SyncRunner()
self.image.setup(self.runner)
def testAudioRipCRC(self):
crctask = image.AudioRipCRCTask(self.image)
self.runner.run(crctask, verbose=False)
def testAccurateRipChecksum(self):
checksumtask = image.AccurateRipChecksumTask(self.image)
self.runner.run(checksumtask, verbose=False)
self.assertEquals(len(crctask.crcs), 4)
self.assertEquals(h(crctask.crcs[0]), '0x00000000')
self.assertEquals(h(crctask.crcs[1]), '0x793fa868')
self.assertEquals(h(crctask.crcs[2]), '0x8dd37c26')
self.assertEquals(h(crctask.crcs[3]), '0x00000000')
self.assertEquals(len(checksumtask.checksums), 4)
self.assertEquals(h(checksumtask.checksums[0]), '0x00000000')
self.assertEquals(h(checksumtask.checksums[1]), '0x793fa868')
self.assertEquals(h(checksumtask.checksums[2]), '0x8dd37c26')
self.assertEquals(h(checksumtask.checksums[3]), '0x00000000')
def testLength(self):
self.assertEquals(self.image.toc.getTrackLength(1), 2)
@@ -50,15 +50,15 @@ class TracSeparateTestCase(unittest.TestCase):
self.runner = task.SyncRunner()
self.image.setup(self.runner)
def testAudioRipCRC(self):
crctask = image.AudioRipCRCTask(self.image)
self.runner.run(crctask, verbose=False)
def testAccurateRipChecksum(self):
checksumtask = image.AccurateRipChecksumTask(self.image)
self.runner.run(checksumtask, verbose=False)
self.assertEquals(len(crctask.crcs), 4)
self.assertEquals(h(crctask.crcs[0]), '0xd60e55e1')
self.assertEquals(h(crctask.crcs[1]), '0xd63dc2d2')
self.assertEquals(h(crctask.crcs[2]), '0xd63dc2d2')
self.assertEquals(h(crctask.crcs[3]), '0x7271db39')
self.assertEquals(len(checksumtask.checksums), 4)
self.assertEquals(h(checksumtask.checksums[0]), '0xd60e55e1')
self.assertEquals(h(checksumtask.checksums[1]), '0xd63dc2d2')
self.assertEquals(h(checksumtask.checksums[2]), '0xd63dc2d2')
self.assertEquals(h(checksumtask.checksums[3]), '0x7271db39')
def testLength(self):
tracks = self.image.cue.tracks
@@ -101,5 +101,5 @@ class AccurateRipResponseTestCase(unittest.TestCase):
for i in range(11):
self.assertEquals(response.confidences[i], 35)
self.assertEquals(response.crcs[0], "beea32c8")
self.assertEquals(response.crcs[10], "acee98ca")
self.assertEquals(response.checksums[0], "beea32c8")
self.assertEquals(response.checksums[10], "acee98ca")