--- /dev/null
+#! /usr/bin/env python
+#
+# mutagen aims to be an all purpose media tagging library
+# Copyright (C) 2005 Michael Urman
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+#
+# $Id: __init__.py 4275 2008-06-01 06:32:37Z piman $
+#
+
+"""Mutagen aims to be an all purpose tagging library.
+
+ import mutagen.[format]
+ metadata = mutagen.[format].Open(filename)
+
+metadata acts like a dictionary of tags in the file. Tags are generally a
+list of string-like values, but may have additional methods available
+depending on tag or format. They may also be entirely different objects
+for certain keys, again depending on format.
+"""
+
+version = (1, 14)
+version_string = ".".join(map(str, version))
+
+import warnings
+
+import mutagen._util
+
+class Metadata(object):
+ """An abstract dict-like object.
+
+ Metadata is the base class for many of the tag objects in Mutagen.
+ """
+
+ def __init__(self, *args, **kwargs):
+ if args or kwargs:
+ self.load(*args, **kwargs)
+
+ def load(self, *args, **kwargs):
+ raise NotImplementedError
+
+ def save(self, filename=None):
+ raise NotImplementedError
+
+ def delete(self, filename=None):
+ raise NotImplementedError
+
+class FileType(mutagen._util.DictMixin):
+ """An abstract object wrapping tags and audio stream information.
+
+ Attributes:
+ info -- stream information (length, bitrate, sample rate)
+ tags -- metadata tags, if any
+
+ Each file format has different potential tags and stream
+ information.
+
+ FileTypes implement an interface very similar to Metadata; the
+ dict interface, save, load, and delete calls on a FileType call
+ the appropriate methods on its tag data.
+ """
+
+ info = None
+ tags = None
+ filename = None
+ _mimes = ["application/octet-stream"]
+
+ def __init__(self, filename=None, *args, **kwargs):
+ if filename is None:
+ warnings.warn("FileType constructor requires a filename",
+ DeprecationWarning)
+ else:
+ self.load(filename, *args, **kwargs)
+
+ def load(self, filename, *args, **kwargs):
+ raise NotImplementedError
+
+ def __getitem__(self, key):
+ """Look up a metadata tag key.
+
+ If the file has no tags at all, a KeyError is raised.
+ """
+ if self.tags is None: raise KeyError, key
+ else: return self.tags[key]
+
+ def __setitem__(self, key, value):
+ """Set a metadata tag.
+
+ If the file has no tags, an appropriate format is added (but
+ not written until save is called).
+ """
+ if self.tags is None:
+ self.add_tags()
+ self.tags[key] = value
+
+ def __delitem__(self, key):
+ """Delete a metadata tag key.
+
+ If the file has no tags at all, a KeyError is raised.
+ """
+ if self.tags is None: raise KeyError, key
+ else: del(self.tags[key])
+
+ def keys(self):
+ """Return a list of keys in the metadata tag.
+
+ If the file has no tags at all, an empty list is returned.
+ """
+ if self.tags is None: return []
+ else: return self.tags.keys()
+
+ def delete(self, filename=None):
+ """Remove tags from a file."""
+ if self.tags is not None:
+ if filename is None:
+ filename = self.filename
+ else:
+ warnings.warn(
+ "delete(filename=...) is deprecated, reload the file",
+ DeprecationWarning)
+ return self.tags.delete(filename)
+
+ def save(self, filename=None, **kwargs):
+ """Save metadata tags."""
+ if filename is None:
+ filename = self.filename
+ else:
+ warnings.warn(
+ "save(filename=...) is deprecated, reload the file",
+ DeprecationWarning)
+ if self.tags is not None:
+ return self.tags.save(filename, **kwargs)
+ else: raise ValueError("no tags in file")
+
+ def pprint(self):
+ """Print stream information and comment key=value pairs."""
+ stream = "%s (%s)" % (self.info.pprint(), self.mime[0])
+ try: tags = self.tags.pprint()
+ except AttributeError:
+ return stream
+ else: return stream + ((tags and "\n" + tags) or "")
+
+ def add_tags(self):
+ raise NotImplementedError
+
+ def __get_mime(self):
+ mimes = []
+ for Kind in type(self).__mro__:
+ for mime in getattr(Kind, '_mimes', []):
+ if mime not in mimes:
+ mimes.append(mime)
+ return mimes
+
+ mime = property(__get_mime)
+
+def File(filename, options=None):
+ """Guess the type of the file and try to open it.
+
+ The file type is decided by several things, such as the first 128
+ bytes (which usually contains a file type identifier), the
+ filename extension, and the presence of existing tags.
+
+ If no appropriate type could be found, None is returned.
+ """
+
+ if options is None:
+ from mutagen.asf import ASF
+ from mutagen.apev2 import APEv2File
+ from mutagen.flac import FLAC
+ from mutagen.id3 import ID3FileType
+ from mutagen.mp3 import MP3
+ from mutagen.oggflac import OggFLAC
+ from mutagen.oggspeex import OggSpeex
+ from mutagen.oggtheora import OggTheora
+ from mutagen.oggvorbis import OggVorbis
+ from mutagen.trueaudio import TrueAudio
+ from mutagen.wavpack import WavPack
+ from mutagen.mp4 import MP4
+ from mutagen.musepack import Musepack
+ from mutagen.monkeysaudio import MonkeysAudio
+ from mutagen.optimfrog import OptimFROG
+ options = [MP3, TrueAudio, OggTheora, OggSpeex, OggVorbis, OggFLAC,
+ FLAC, APEv2File, MP4, ID3FileType, WavPack, Musepack,
+ MonkeysAudio, OptimFROG, ASF]
+
+ if not options:
+ return None
+
+ fileobj = file(filename, "rb")
+ try:
+ header = fileobj.read(128)
+ results = [Kind.score(filename, fileobj, header) for Kind in options]
+ finally:
+ fileobj.close()
+ results = zip(results, options)
+ results.sort()
+ score, Kind = results[-1]
+ if score > 0: return Kind(filename)
+ else: return None
--- /dev/null
+"""Constants used by Mutagen."""
+
+GENRES = [
+ u"Blues",
+ u"Classic Rock",
+ u"Country",
+ u"Dance",
+ u"Disco",
+ u"Funk",
+ u"Grunge",
+ u"Hip-Hop",
+ u"Jazz",
+ u"Metal",
+ u"New Age",
+ u"Oldies",
+ u"Other",
+ u"Pop",
+ u"R&B",
+ u"Rap",
+ u"Reggae",
+ u"Rock",
+ u"Techno",
+ u"Industrial",
+ u"Alternative",
+ u"Ska",
+ u"Death Metal",
+ u"Pranks",
+ u"Soundtrack",
+ u"Euro-Techno",
+ u"Ambient",
+ u"Trip-Hop",
+ u"Vocal",
+ u"Jazz+Funk",
+ u"Fusion",
+ u"Trance",
+ u"Classical",
+ u"Instrumental",
+ u"Acid",
+ u"House",
+ u"Game",
+ u"Sound Clip",
+ u"Gospel",
+ u"Noise",
+ u"Alt. Rock",
+ u"Bass",
+ u"Soul",
+ u"Punk",
+ u"Space",
+ u"Meditative",
+ u"Instrumental Pop",
+ u"Instrumental Rock",
+ u"Ethnic",
+ u"Gothic",
+ u"Darkwave",
+ u"Techno-Industrial",
+ u"Electronic",
+ u"Pop-Folk",
+ u"Eurodance",
+ u"Dream",
+ u"Southern Rock",
+ u"Comedy",
+ u"Cult",
+ u"Gangsta",
+ u"Top 40",
+ u"Christian Rap",
+ u"Pop/Funk",
+ u"Jungle",
+ u"Native American",
+ u"Cabaret",
+ u"New Wave",
+ u"Psychadelic",
+ u"Rave",
+ u"Showtunes",
+ u"Trailer",
+ u"Lo-Fi",
+ u"Tribal",
+ u"Acid Punk",
+ u"Acid Jazz",
+ u"Polka",
+ u"Retro",
+ u"Musical",
+ u"Rock & Roll",
+ u"Hard Rock",
+ u"Folk",
+ u"Folk/Rock",
+ u"National Folk",
+ u"Swing",
+ u"Fusion",
+ u"Bebob",
+ u"Latin",
+ u"Revival",
+ u"Celtic",
+ u"Bluegrass",
+ u"Avantgarde",
+ u"Gothic Rock",
+ u"Progressive Rock",
+ u"Psychadelic Rock",
+ u"Symphonic Rock",
+ u"Slow Rock",
+ u"Big Band",
+ u"Chorus",
+ u"Easy Listening",
+ u"Acoustic",
+ u"Humour",
+ u"Speech",
+ u"Chanson",
+ u"Opera",
+ u"Chamber Music",
+ u"Sonata",
+ u"Symphony",
+ u"Booty Bass",
+ u"Primus",
+ u"Porn Groove",
+ u"Satire",
+ u"Slow Jam",
+ u"Club",
+ u"Tango",
+ u"Samba",
+ u"Folklore",
+ u"Ballad",
+ u"Power Ballad",
+ u"Rhythmic Soul",
+ u"Freestyle",
+ u"Duet",
+ u"Punk Rock",
+ u"Drum Solo",
+ u"A Capella",
+ u"Euro-House",
+ u"Dance Hall",
+ u"Goa",
+ u"Drum & Bass",
+ u"Club-House",
+ u"Hardcore",
+ u"Terror",
+ u"Indie",
+ u"BritPop",
+ u"Negerpunk",
+ u"Polsk Punk",
+ u"Beat",
+ u"Christian Gangsta Rap",
+ u"Heavy Metal",
+ u"Black Metal",
+ u"Crossover",
+ u"Contemporary Christian",
+ u"Christian Rock",
+ u"Merengue",
+ u"Salsa",
+ u"Thrash Metal",
+ u"Anime",
+ u"Jpop",
+ u"Synthpop"
+ ]
+"""The ID3v1 genre list."""
--- /dev/null
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: _util.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Utility classes for Mutagen.
+
+You should not rely on the interfaces here being stable. They are
+intended for internal use in Mutagen only.
+"""
+
+import struct
+
+class DictMixin(object):
+ """Implement the dict API using keys() and __*item__ methods.
+
+ Similar to UserDict.DictMixin, this takes a class that defines
+ __getitem__, __setitem__, __delitem__, and keys(), and turns it
+ into a full dict-like object.
+
+ UserDict.DictMixin is not suitable for this purpose because it's
+ an old-style class.
+
+ This class is not optimized for very large dictionaries; many
+ functions have linear memory requirements. I recommend you
+ override some of these functions if speed is required.
+ """
+
+ def __iter__(self):
+ return iter(self.keys())
+
+ def has_key(self, key):
+ try: self[key]
+ except KeyError: return False
+ else: return True
+ __contains__ = has_key
+
+ iterkeys = lambda self: iter(self.keys())
+
+ def values(self):
+ return map(self.__getitem__, self.keys())
+ itervalues = lambda self: iter(self.values())
+
+ def items(self):
+ return zip(self.keys(), self.values())
+ iteritems = lambda s: iter(s.items())
+
+ def clear(self):
+ map(self.__delitem__, self.keys())
+
+ def pop(self, key, *args):
+ if len(args) > 1:
+ raise TypeError("pop takes at most two arguments")
+ try: value = self[key]
+ except KeyError:
+ if args: return args[0]
+ else: raise
+ del(self[key])
+ return value
+
+ def popitem(self):
+ try:
+ key = self.keys()[0]
+ return key, self.pop(key)
+ except IndexError: raise KeyError("dictionary is empty")
+
+ def update(self, other=None, **kwargs):
+ if other is None:
+ self.update(kwargs)
+ other = {}
+
+ try: map(self.__setitem__, other.keys(), other.values())
+ except AttributeError:
+ for key, value in other:
+ self[key] = value
+
+ def setdefault(self, key, default=None):
+ try: return self[key]
+ except KeyError:
+ self[key] = default
+ return default
+
+ def get(self, key, default=None):
+ try: return self[key]
+ except KeyError: return default
+
+ def __repr__(self):
+ return repr(dict(self.items()))
+
+ def __cmp__(self, other):
+ if other is None: return 1
+ else: return cmp(dict(self.items()), other)
+
+ def __len__(self):
+ return len(self.keys())
+
+class DictProxy(DictMixin):
+ def __init__(self, *args, **kwargs):
+ self.__dict = {}
+ super(DictProxy, self).__init__(*args, **kwargs)
+
+ def __getitem__(self, key):
+ return self.__dict[key]
+
+ def __setitem__(self, key, value):
+ self.__dict[key] = value
+
+ def __delitem__(self, key):
+ del(self.__dict[key])
+
+ def keys(self):
+ return self.__dict.keys()
+
+class cdata(object):
+ """C character buffer to Python numeric type conversions."""
+
+ from struct import error
+
+ short_le = staticmethod(lambda data: struct.unpack('<h', data)[0])
+ ushort_le = staticmethod(lambda data: struct.unpack('<H', data)[0])
+
+ short_be = staticmethod(lambda data: struct.unpack('>h', data)[0])
+ ushort_be = staticmethod(lambda data: struct.unpack('>H', data)[0])
+
+ int_le = staticmethod(lambda data: struct.unpack('<i', data)[0])
+ uint_le = staticmethod(lambda data: struct.unpack('<I', data)[0])
+
+ int_be = staticmethod(lambda data: struct.unpack('>i', data)[0])
+ uint_be = staticmethod(lambda data: struct.unpack('>I', data)[0])
+
+ longlong_le = staticmethod(lambda data: struct.unpack('<q', data)[0])
+ ulonglong_le = staticmethod(lambda data: struct.unpack('<Q', data)[0])
+
+ longlong_be = staticmethod(lambda data: struct.unpack('>q', data)[0])
+ ulonglong_be = staticmethod(lambda data: struct.unpack('>Q', data)[0])
+
+ to_short_le = staticmethod(lambda data: struct.pack('<h', data))
+ to_ushort_le = staticmethod(lambda data: struct.pack('<H', data))
+
+ to_short_be = staticmethod(lambda data: struct.pack('>h', data))
+ to_ushort_be = staticmethod(lambda data: struct.pack('>H', data))
+
+ to_int_le = staticmethod(lambda data: struct.pack('<i', data))
+ to_uint_le = staticmethod(lambda data: struct.pack('<I', data))
+
+ to_int_be = staticmethod(lambda data: struct.pack('>i', data))
+ to_uint_be = staticmethod(lambda data: struct.pack('>I', data))
+
+ to_longlong_le = staticmethod(lambda data: struct.pack('<q', data))
+ to_ulonglong_le = staticmethod(lambda data: struct.pack('<Q', data))
+
+ to_longlong_be = staticmethod(lambda data: struct.pack('>q', data))
+ to_ulonglong_be = staticmethod(lambda data: struct.pack('>Q', data))
+
+ bitswap = ''.join([chr(sum([((val >> i) & 1) << (7-i) for i in range(8)]))
+ for val in range(256)])
+ del(i)
+ del(val)
+
+ test_bit = staticmethod(lambda value, n: bool((value >> n) & 1))
+
+def lock(fileobj):
+ """Lock a file object 'safely'.
+
+ That means a failure to lock because the platform doesn't
+ support fcntl or filesystem locks is not considered a
+ failure. This call does block.
+
+ Returns whether or not the lock was successful, or
+ raises an exception in more extreme circumstances (full
+ lock table, invalid file).
+ """
+ try: import fcntl
+ except ImportError:
+ return False
+ else:
+ try: fcntl.lockf(fileobj, fcntl.LOCK_EX)
+ except IOError:
+ # FIXME: There's possibly a lot of complicated
+ # logic that needs to go here in case the IOError
+ # is EACCES or EAGAIN.
+ return False
+ else:
+ return True
+
+def unlock(fileobj):
+ """Unlock a file object.
+
+ Don't call this on a file object unless a call to lock()
+ returned true.
+ """
+ # If this fails there's a mismatched lock/unlock pair,
+ # so we definitely don't want to ignore errors.
+ import fcntl
+ fcntl.lockf(fileobj, fcntl.LOCK_UN)
+
+def insert_bytes(fobj, size, offset, BUFFER_SIZE=2**16):
+ """Insert size bytes of empty space starting at offset.
+
+ fobj must be an open file object, open rb+ or
+ equivalent. Mutagen tries to use mmap to resize the file, but
+ falls back to a significantly slower method if mmap fails.
+ """
+ assert 0 < size
+ assert 0 <= offset
+ locked = False
+ fobj.seek(0, 2)
+ filesize = fobj.tell()
+ movesize = filesize - offset
+ fobj.write('\x00' * size)
+ fobj.flush()
+ try:
+ try:
+ import mmap
+ map = mmap.mmap(fobj.fileno(), filesize + size)
+ try: map.move(offset + size, offset, movesize)
+ finally: map.close()
+ except (ValueError, EnvironmentError, ImportError):
+ # handle broken mmap scenarios
+ locked = lock(fobj)
+ fobj.truncate(filesize)
+
+ fobj.seek(0, 2)
+ padsize = size
+ # Don't generate an enormous string if we need to pad
+ # the file out several megs.
+ while padsize:
+ addsize = min(BUFFER_SIZE, padsize)
+ fobj.write("\x00" * addsize)
+ padsize -= addsize
+
+ fobj.seek(filesize, 0)
+ while movesize:
+ # At the start of this loop, fobj is pointing at the end
+ # of the data we need to move, which is of movesize length.
+ thismove = min(BUFFER_SIZE, movesize)
+ # Seek back however much we're going to read this frame.
+ fobj.seek(-thismove, 1)
+ nextpos = fobj.tell()
+ # Read it, so we're back at the end.
+ data = fobj.read(thismove)
+ # Seek back to where we need to write it.
+ fobj.seek(-thismove + size, 1)
+ # Write it.
+ fobj.write(data)
+ # And seek back to the end of the unmoved data.
+ fobj.seek(nextpos)
+ movesize -= thismove
+
+ fobj.flush()
+ finally:
+ if locked:
+ unlock(fobj)
+
+def delete_bytes(fobj, size, offset, BUFFER_SIZE=2**16):
+ """Delete size bytes of empty space starting at offset.
+
+ fobj must be an open file object, open rb+ or
+ equivalent. Mutagen tries to use mmap to resize the file, but
+ falls back to a significantly slower method if mmap fails.
+ """
+ locked = False
+ assert 0 < size
+ assert 0 <= offset
+ fobj.seek(0, 2)
+ filesize = fobj.tell()
+ movesize = filesize - offset - size
+ assert 0 <= movesize
+ try:
+ if movesize > 0:
+ fobj.flush()
+ try:
+ import mmap
+ map = mmap.mmap(fobj.fileno(), filesize)
+ try: map.move(offset, offset + size, movesize)
+ finally: map.close()
+ except (ValueError, EnvironmentError, ImportError):
+ # handle broken mmap scenarios
+ locked = lock(fobj)
+ fobj.seek(offset + size)
+ buf = fobj.read(BUFFER_SIZE)
+ while buf:
+ fobj.seek(offset)
+ fobj.write(buf)
+ offset += len(buf)
+ fobj.seek(offset + size)
+ buf = fobj.read(BUFFER_SIZE)
+ fobj.truncate(filesize - size)
+ fobj.flush()
+ finally:
+ if locked:
+ unlock(fobj)
+
+def utf8(data):
+ """Convert a basestring to a valid UTF-8 str."""
+ if isinstance(data, str):
+ return data.decode("utf-8", "replace").encode("utf-8")
+ elif isinstance(data, unicode):
+ return data.encode("utf-8")
+ else: raise TypeError("only unicode/str types can be converted to UTF-8")
--- /dev/null
+# Vorbis comment support for Mutagen
+# Copyright 2005-2006 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+
+"""Read and write Vorbis comment data.
+
+Vorbis comments are freeform key/value pairs; keys are
+case-insensitive ASCII and values are Unicode strings. A key may have
+multiple values.
+
+The specification is at http://www.xiph.org/vorbis/doc/v-comment.html.
+"""
+
+import sys
+
+from cStringIO import StringIO
+
+import mutagen
+from mutagen._util import DictMixin, cdata
+
+try: set
+except NameError:
+ from sets import Set as set
+
+def is_valid_key(key):
+ """Return true if a string is a valid Vorbis comment key.
+
+ Valid Vorbis comment keys are printable ASCII between 0x20 (space)
+ and 0x7D ('}'), excluding '='.
+ """
+ for c in key:
+ if c < " " or c > "}" or c == "=": return False
+ else: return bool(key)
+istag = is_valid_key
+
+class error(IOError): pass
+class VorbisUnsetFrameError(error): pass
+class VorbisEncodingError(error): pass
+
+class VComment(mutagen.Metadata, list):
+ """A Vorbis comment parser, accessor, and renderer.
+
+ All comment ordering is preserved. A VComment is a list of
+ key/value pairs, and so any Python list method can be used on it.
+
+ Vorbis comments are always wrapped in something like an Ogg Vorbis
+ bitstream or a FLAC metadata block, so this loads string data or a
+ file-like object, not a filename.
+
+ Attributes:
+ vendor -- the stream 'vendor' (i.e. writer); default 'Mutagen'
+ """
+
+ vendor = u"Mutagen " + mutagen.version_string
+
+ def __init__(self, data=None, *args, **kwargs):
+ # Collect the args to pass to load, this lets child classes
+ # override just load and get equivalent magic for the
+ # constructor.
+ if data is not None:
+ if isinstance(data, str):
+ data = StringIO(data)
+ elif not hasattr(data, 'read'):
+ raise TypeError("VComment requires string data or a file-like")
+ self.load(data, *args, **kwargs)
+
+ def load(self, fileobj, errors='replace', framing=True):
+ """Parse a Vorbis comment from a file-like object.
+
+ Keyword arguments:
+ errors:
+ 'strict', 'replace', or 'ignore'. This affects Unicode decoding
+ and how other malformed content is interpreted.
+ framing -- if true, fail if a framing bit is not present
+
+ Framing bits are required by the Vorbis comment specification,
+ but are not used in FLAC Vorbis comment blocks.
+
+ """
+ try:
+ vendor_length = cdata.uint_le(fileobj.read(4))
+ self.vendor = fileobj.read(vendor_length).decode('utf-8', errors)
+ count = cdata.uint_le(fileobj.read(4))
+ for i in range(count):
+ length = cdata.uint_le(fileobj.read(4))
+ try: string = fileobj.read(length).decode('utf-8', errors)
+ except (OverflowError, MemoryError):
+ raise error("cannot read %d bytes, too large" % length)
+ try: tag, value = string.split('=', 1)
+ except ValueError, err:
+ if errors == "ignore":
+ continue
+ elif errors == "replace":
+ tag, value = u"unknown%d" % i, string
+ else:
+ raise VorbisEncodingError, str(err), sys.exc_info()[2]
+ try: tag = tag.encode('ascii', errors)
+ except UnicodeEncodeError:
+ raise VorbisEncodingError, "invalid tag name %r" % tag
+ else:
+ if is_valid_key(tag): self.append((tag, value))
+ if framing and not ord(fileobj.read(1)) & 0x01:
+ raise VorbisUnsetFrameError("framing bit was unset")
+ except (cdata.error, TypeError):
+ raise error("file is not a valid Vorbis comment")
+
+ def validate(self):
+ """Validate keys and values.
+
+ Check to make sure every key used is a valid Vorbis key, and
+ that every value used is a valid Unicode or UTF-8 string. If
+ any invalid keys or values are found, a ValueError is raised.
+ """
+
+ if not isinstance(self.vendor, unicode):
+ try: self.vendor.decode('utf-8')
+ except UnicodeDecodeError: raise ValueError
+
+ for key, value in self:
+ try:
+ if not is_valid_key(key): raise ValueError
+ except: raise ValueError("%r is not a valid key" % key)
+ if not isinstance(value, unicode):
+ try: value.encode("utf-8")
+ except: raise ValueError("%r is not a valid value" % value)
+ else: return True
+
+ def clear(self):
+ """Clear all keys from the comment."""
+ del(self[:])
+
+ def write(self, framing=True):
+ """Return a string representation of the data.
+
+ Validation is always performed, so calling this function on
+ invalid data may raise a ValueError.
+
+ Keyword arguments:
+ framing -- if true, append a framing bit (see load)
+ """
+
+ self.validate()
+
+ f = StringIO()
+ f.write(cdata.to_uint_le(len(self.vendor.encode('utf-8'))))
+ f.write(self.vendor.encode('utf-8'))
+ f.write(cdata.to_uint_le(len(self)))
+ for tag, value in self:
+ comment = "%s=%s" % (tag, value.encode('utf-8'))
+ f.write(cdata.to_uint_le(len(comment)))
+ f.write(comment)
+ if framing: f.write("\x01")
+ return f.getvalue()
+
+ def pprint(self):
+ return "\n".join(["%s=%s" % (k.lower(), v) for k, v in self])
+
+class VCommentDict(VComment, DictMixin):
+ """A VComment that looks like a dictionary.
+
+ This object differs from a dictionary in two ways. First,
+ len(comment) will still return the number of values, not the
+ number of keys. Secondly, iterating through the object will
+ iterate over (key, value) pairs, not keys. Since a key may have
+ multiple values, the same value may appear multiple times while
+ iterating.
+
+ Since Vorbis comment keys are case-insensitive, all keys are
+ normalized to lowercase ASCII.
+ """
+
+ def __getitem__(self, key):
+ """A list of values for the key.
+
+ This is a copy, so comment['title'].append('a title') will not
+ work.
+
+ """
+ key = key.lower().encode('ascii')
+ values = [value for (k, value) in self if k.lower() == key]
+ if not values: raise KeyError, key
+ else: return values
+
+ def __delitem__(self, key):
+ """Delete all values associated with the key."""
+ key = key.lower().encode('ascii')
+ to_delete = filter(lambda x: x[0].lower() == key, self)
+ if not to_delete:raise KeyError, key
+ else: map(self.remove, to_delete)
+
+ def __contains__(self, key):
+ """Return true if the key has any values."""
+ key = key.lower().encode('ascii')
+ for k, value in self:
+ if k.lower() == key: return True
+ else: return False
+
+ def __setitem__(self, key, values):
+ """Set a key's value or values.
+
+ Setting a value overwrites all old ones. The value may be a
+ list of Unicode or UTF-8 strings, or a single Unicode or UTF-8
+ string.
+
+ """
+ key = key.lower().encode('ascii')
+ if not isinstance(values, list):
+ values = [values]
+ try: del(self[key])
+ except KeyError: pass
+ for value in values:
+ self.append((key, value))
+
+ def keys(self):
+ """Return all keys in the comment."""
+ return self and map(str.lower, set(zip(*self)[0]))
+
+ def as_dict(self):
+ """Return a copy of the comment data in a real dict."""
+ d = {}
+ for key, value in self:
+ d.setdefault(key, []).append(value)
+ return d
--- /dev/null
+# An APEv2 tag reader
+#
+# Copyright 2005 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: apev2.py 4275 2008-06-01 06:32:37Z piman $
+
+"""APEv2 reading and writing.
+
+The APEv2 format is most commonly used with Musepack files, but is
+also the format of choice for WavPack and other formats. Some MP3s
+also have APEv2 tags, but this can cause problems with many MP3
+decoders and taggers.
+
+APEv2 tags, like Vorbis comments, are freeform key=value pairs. APEv2
+keys can be any ASCII string with characters from 0x20 to 0x7E,
+between 2 and 255 characters long. Keys are case-sensitive, but
+readers are recommended to be case insensitive, and it is forbidden to
+multiple keys which differ only in case. Keys are usually stored
+title-cased (e.g. 'Artist' rather than 'artist').
+
+APEv2 values are slightly more structured than Vorbis comments; values
+are flagged as one of text, binary, or an external reference (usually
+a URI).
+
+Based off the format specification found at
+http://wiki.hydrogenaudio.org/index.php?title=APEv2_specification.
+"""
+
+__all__ = ["APEv2", "APEv2File", "Open", "delete"]
+
+import struct
+from cStringIO import StringIO
+
+def is_valid_apev2_key(key):
+ return (2 <= len(key) <= 255 and min(key) >= ' ' and max(key) <= '~' and
+ key not in ["OggS", "TAG", "ID3", "MP+"])
+
+# There are three different kinds of APE tag values.
+# "0: Item contains text information coded in UTF-8
+# 1: Item contains binary information
+# 2: Item is a locator of external stored information [e.g. URL]
+# 3: reserved"
+TEXT, BINARY, EXTERNAL = range(3)
+
+HAS_HEADER = 1L << 31
+HAS_NO_FOOTER = 1L << 30
+IS_HEADER = 1L << 29
+
+class error(IOError): pass
+class APENoHeaderError(error, ValueError): pass
+class APEUnsupportedVersionError(error, ValueError): pass
+class APEBadItemError(error, ValueError): pass
+
+from mutagen import Metadata, FileType
+from mutagen._util import DictMixin, cdata, utf8, delete_bytes
+
+class _APEv2Data(object):
+ # Store offsets of the important parts of the file.
+ start = header = data = footer = end = None
+ # Footer or header; seek here and read 32 to get version/size/items/flags
+ metadata = None
+ # Actual tag data
+ tag = None
+
+ version = None
+ size = None
+ items = None
+ flags = 0
+
+ # The tag is at the start rather than the end. A tag at both
+ # the start and end of the file (i.e. the tag is the whole file)
+ # is not considered to be at the start.
+ is_at_start = False
+
+ def __init__(self, fileobj):
+ self.__find_metadata(fileobj)
+ self.metadata = max(self.header, self.footer)
+ if self.metadata is None: return
+ self.__fill_missing(fileobj)
+ self.__fix_brokenness(fileobj)
+ if self.data is not None:
+ fileobj.seek(self.data)
+ self.tag = fileobj.read(self.size)
+
+ def __find_metadata(self, fileobj):
+ # Try to find a header or footer.
+
+ # Check for a simple footer.
+ try: fileobj.seek(-32, 2)
+ except IOError:
+ fileobj.seek(0, 2)
+ return
+ if fileobj.read(8) == "APETAGEX":
+ fileobj.seek(-8, 1)
+ self.footer = self.metadata = fileobj.tell()
+ return
+
+ # Check for an APEv2 tag followed by an ID3v1 tag at the end.
+ try:
+ fileobj.seek(-128, 2)
+ if fileobj.read(3) == "TAG":
+
+ fileobj.seek(-35, 1) # "TAG" + header length
+ if fileobj.read(8) == "APETAGEX":
+ fileobj.seek(-8, 1)
+ self.footer = fileobj.tell()
+ return
+
+ # ID3v1 tag at the end, maybe preceded by Lyrics3v2.
+ # (http://www.id3.org/lyrics3200.html)
+ # (header length - "APETAGEX") - "LYRICS200"
+ fileobj.seek(15, 1)
+ if fileobj.read(9) == 'LYRICS200':
+ fileobj.seek(-15, 1) # "LYRICS200" + size tag
+ try: offset = int(fileobj.read(6))
+ except ValueError:
+ raise IOError
+
+ fileobj.seek(-32 - offset - 6, 1)
+ if fileobj.read(8) == "APETAGEX":
+ fileobj.seek(-8, 1)
+ self.footer = fileobj.tell()
+ return
+
+ except IOError:
+ pass
+
+ # Check for a tag at the start.
+ fileobj.seek(0, 0)
+ if fileobj.read(8) == "APETAGEX":
+ self.is_at_start = True
+ self.header = 0
+
+ def __fill_missing(self, fileobj):
+ fileobj.seek(self.metadata + 8)
+ self.version = fileobj.read(4)
+ self.size = cdata.uint_le(fileobj.read(4))
+ self.items = cdata.uint_le(fileobj.read(4))
+ self.flags = cdata.uint_le(fileobj.read(4))
+
+ if self.header is not None:
+ self.data = self.header + 32
+ # If we're reading the header, the size is the header
+ # offset + the size, which includes the footer.
+ self.end = self.data + self.size
+ fileobj.seek(self.end - 32, 0)
+ if fileobj.read(8) == "APETAGEX":
+ self.footer = self.end - 32
+ elif self.footer is not None:
+ self.end = self.footer + 32
+ self.data = self.end - self.size
+ if self.flags & HAS_HEADER:
+ self.header = self.data - 32
+ else:
+ self.header = self.data
+ else: raise APENoHeaderError("No APE tag found")
+
+ def __fix_brokenness(self, fileobj):
+ # Fix broken tags written with PyMusepack.
+ if self.header is not None: start = self.header
+ else: start = self.data
+ fileobj.seek(start)
+
+ while start > 0:
+ # Clean up broken writing from pre-Mutagen PyMusepack.
+ # It didn't remove the first 24 bytes of header.
+ try: fileobj.seek(-24, 1)
+ except IOError:
+ break
+ else:
+ if fileobj.read(8) == "APETAGEX":
+ fileobj.seek(-8, 1)
+ start = fileobj.tell()
+ else: break
+ self.start = start
+
+class APEv2(DictMixin, Metadata):
+ """A file with an APEv2 tag.
+
+ ID3v1 tags are silently ignored and overwritten.
+ """
+
+ filename = None
+
+ def __init__(self, *args, **kwargs):
+ self.__casemap = {}
+ self.__dict = {}
+ super(APEv2, self).__init__(*args, **kwargs)
+ # Internally all names are stored as lowercase, but the case
+ # they were set with is remembered and used when saving. This
+ # is roughly in line with the standard, which says that keys
+ # are case-sensitive but two keys differing only in case are
+ # not allowed, and recommends case-insensitive
+ # implementations.
+
+ def pprint(self):
+ """Return tag key=value pairs in a human-readable format."""
+ items = self.items()
+ items.sort()
+ return "\n".join(["%s=%s" % (k, v.pprint()) for k, v in items])
+
+ def load(self, filename):
+ """Load tags from a filename."""
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ data = _APEv2Data(fileobj)
+ finally:
+ fileobj.close()
+ if data.tag:
+ self.clear()
+ self.__casemap.clear()
+ self.__parse_tag(data.tag, data.items)
+ else:
+ raise APENoHeaderError("No APE tag found")
+
+ def __parse_tag(self, tag, count):
+ fileobj = StringIO(tag)
+
+ for i in range(count):
+ size = cdata.uint_le(fileobj.read(4))
+ flags = cdata.uint_le(fileobj.read(4))
+
+ # Bits 1 and 2 bits are flags, 0-3
+ # Bit 0 is read/write flag, ignored
+ kind = (flags & 6) >> 1
+ if kind == 3:
+ raise APEBadItemError("value type must be 0, 1, or 2")
+ key = value = fileobj.read(1)
+ while key[-1:] != '\x00' and value:
+ value = fileobj.read(1)
+ key += value
+ if key[-1:] == "\x00":
+ key = key[:-1]
+ value = fileobj.read(size)
+ self[key] = APEValue(value, kind)
+
+ def __getitem__(self, key):
+ if not is_valid_apev2_key(key):
+ raise KeyError("%r is not a valid APEv2 key" % key)
+ return self.__dict[key.lower()]
+
+ def __delitem__(self, key):
+ if not is_valid_apev2_key(key):
+ raise KeyError("%r is not a valid APEv2 key" % key)
+ del(self.__dict[key.lower()])
+
+ def __setitem__(self, key, value):
+ """'Magic' value setter.
+
+ This function tries to guess at what kind of value you want to
+ store. If you pass in a valid UTF-8 or Unicode string, it
+ treats it as a text value. If you pass in a list, it treats it
+ as a list of string/Unicode values. If you pass in a string
+ that is not valid UTF-8, it assumes it is a binary value.
+
+ If you need to force a specific type of value (e.g. binary
+ data that also happens to be valid UTF-8, or an external
+ reference), use the APEValue factory and set the value to the
+ result of that:
+ from mutagen.apev2 import APEValue, EXTERNAL
+ tag['Website'] = APEValue('http://example.org', EXTERNAL)
+ """
+
+ if not is_valid_apev2_key(key):
+ raise KeyError("%r is not a valid APEv2 key" % key)
+
+ if not isinstance(value, _APEValue):
+ # let's guess at the content if we're not already a value...
+ if isinstance(value, unicode):
+ # unicode? we've got to be text.
+ value = APEValue(utf8(value), TEXT)
+ elif isinstance(value, list):
+ # list? text.
+ value = APEValue("\0".join(map(utf8, value)), TEXT)
+ else:
+ try: dummy = value.decode("utf-8")
+ except UnicodeError:
+ # invalid UTF8 text, probably binary
+ value = APEValue(value, BINARY)
+ else:
+ # valid UTF8, probably text
+ value = APEValue(value, TEXT)
+ self.__casemap[key.lower()] = key
+ self.__dict[key.lower()] = value
+
+ def keys(self):
+ return [self.__casemap.get(key, key) for key in self.__dict.keys()]
+
+ def save(self, filename=None):
+ """Save changes to a file.
+
+ If no filename is given, the one most recently loaded is used.
+
+ Tags are always written at the end of the file, and include
+ a header and a footer.
+ """
+
+ filename = filename or self.filename
+ try:
+ fileobj = file(filename, "r+b")
+ except IOError:
+ fileobj = file(filename, "w+b")
+ data = _APEv2Data(fileobj)
+
+ if data.is_at_start:
+ delete_bytes(fileobj, data.end - data.start, data.start)
+ elif data.start is not None:
+ fileobj.seek(data.start)
+ # Delete an ID3v1 tag if present, too.
+ fileobj.truncate()
+ fileobj.seek(0, 2)
+
+ # "APE tags items should be sorted ascending by size... This is
+ # not a MUST, but STRONGLY recommended. Actually the items should
+ # be sorted by importance/byte, but this is not feasible."
+ tags = [v._internal(k) for k, v in self.items()]
+ tags.sort(lambda a, b: cmp(len(a), len(b)))
+ num_tags = len(tags)
+ tags = "".join(tags)
+
+ header = "APETAGEX%s%s" %(
+ # version, tag size, item count, flags
+ struct.pack("<4I", 2000, len(tags) + 32, num_tags,
+ HAS_HEADER | IS_HEADER),
+ "\0" * 8)
+ fileobj.write(header)
+
+ fileobj.write(tags)
+
+ footer = "APETAGEX%s%s" %(
+ # version, tag size, item count, flags
+ struct.pack("<4I", 2000, len(tags) + 32, num_tags,
+ HAS_HEADER),
+ "\0" * 8)
+ fileobj.write(footer)
+ fileobj.close()
+
+ def delete(self, filename=None):
+ """Remove tags from a file."""
+ filename = filename or self.filename
+ fileobj = file(filename, "r+b")
+ try:
+ data = _APEv2Data(fileobj)
+ if data.start is not None and data.size is not None:
+ delete_bytes(fileobj, data.end - data.start, data.start)
+ finally:
+ fileobj.close()
+ self.clear()
+
+Open = APEv2
+
+def delete(filename):
+ """Remove tags from a file."""
+ try: APEv2(filename).delete()
+ except APENoHeaderError: pass
+
+def APEValue(value, kind):
+ """APEv2 tag value factory.
+
+ Use this if you need to specify the value's type manually. Binary
+ and text data are automatically detected by APEv2.__setitem__.
+ """
+ if kind == TEXT: return APETextValue(value, kind)
+ elif kind == BINARY: return APEBinaryValue(value, kind)
+ elif kind == EXTERNAL: return APEExtValue(value, kind)
+ else: raise ValueError("kind must be TEXT, BINARY, or EXTERNAL")
+
+class _APEValue(object):
+ def __init__(self, value, kind):
+ self.kind = kind
+ self.value = value
+
+ def __len__(self):
+ return len(self.value)
+ def __str__(self):
+ return self.value
+
+ # Packed format for an item:
+ # 4B: Value length
+ # 4B: Value type
+ # Key name
+ # 1B: Null
+ # Key value
+ def _internal(self, key):
+ return "%s%s\0%s" %(
+ struct.pack("<2I", len(self.value), self.kind << 1),
+ key, self.value)
+
+ def __repr__(self):
+ return "%s(%r, %d)" % (type(self).__name__, self.value, self.kind)
+
+class APETextValue(_APEValue):
+ """An APEv2 text value.
+
+ Text values are Unicode/UTF-8 strings. They can be accessed like
+ strings (with a null seperating the values), or arrays of strings."""
+
+ def __unicode__(self):
+ return unicode(str(self), "utf-8")
+
+ def __iter__(self):
+ """Iterate over the strings of the value (not the characters)"""
+ return iter(unicode(self).split("\0"))
+
+ def __getitem__(self, index):
+ return unicode(self).split("\0")[index]
+
+ def __len__(self):
+ return self.value.count("\0") + 1
+
+ def __cmp__(self, other):
+ return cmp(unicode(self), other)
+
+ def __setitem__(self, index, value):
+ values = list(self)
+ values[index] = value.encode("utf-8")
+ self.value = "\0".join(values).encode("utf-8")
+
+ def pprint(self):
+ return " / ".join(self)
+
+class APEBinaryValue(_APEValue):
+ """An APEv2 binary value."""
+
+ def pprint(self): return "[%d bytes]" % len(self)
+
+class APEExtValue(_APEValue):
+ """An APEv2 external value.
+
+ External values are usually URI or IRI strings.
+ """
+ def pprint(self): return "[External] %s" % unicode(self)
+
+class APEv2File(FileType):
+ class _Info(object):
+ length = 0
+ bitrate = 0
+ def __init__(self, fileobj): pass
+ pprint = staticmethod(lambda: "Unknown format with APEv2 tag.")
+
+ def load(self, filename):
+ self.filename = filename
+ self.info = self._Info(file(filename, "rb"))
+ try: self.tags = APEv2(filename)
+ except error: self.tags = None
+
+ def add_tags(self):
+ if self.tags is None:
+ self.tags = APEv2()
+ else:
+ raise ValueError("%r already has tags: %r" % (self, self.tags))
+
+ def score(filename, fileobj, header):
+ try: fileobj.seek(-160, 2)
+ except IOError:
+ fileobj.seek(0)
+ footer = fileobj.read()
+ filename = filename.lower()
+ return (("APETAGEX" in footer) - header.startswith("ID3"))
+ score = staticmethod(score)
--- /dev/null
+# Copyright 2006-2007 Lukas Lalinsky
+# Copyright 2005-2006 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: asf.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write ASF (Window Media Audio) files."""
+
+__all__ = ["ASF", "Open"]
+
+import struct
+from mutagen import FileType, Metadata
+from mutagen._util import insert_bytes, delete_bytes, DictMixin
+
+class error(IOError): pass
+class ASFError(error): pass
+class ASFHeaderError(error): pass
+
+
+class ASFInfo(object):
+ """ASF stream information."""
+
+ def __init__(self):
+ self.length = 0.0
+ self.sample_rate = 0
+ self.bitrate = 0
+ self.channels = 0
+
+ def pprint(self):
+ s = "Windows Media Audio %d bps, %s Hz, %d channels, %.2f seconds" % (
+ self.bitrate, self.sample_rate, self.channels, self.length)
+ return s
+
+
+class ASFTags(list, DictMixin, Metadata):
+ """Dictionary containing ASF attributes."""
+
+ def pprint(self):
+ return "\n".join(["%s=%s" % (k, v) for k, v in self])
+
+ def __getitem__(self, key):
+ """A list of values for the key.
+
+ This is a copy, so comment['title'].append('a title') will not
+ work.
+
+ """
+ values = [value for (k, value) in self if k == key]
+ if not values: raise KeyError, key
+ else: return values
+
+ def __delitem__(self, key):
+ """Delete all values associated with the key."""
+ to_delete = filter(lambda x: x[0] == key, self)
+ if not to_delete: raise KeyError, key
+ else: map(self.remove, to_delete)
+
+ def __contains__(self, key):
+ """Return true if the key has any values."""
+ for k, value in self:
+ if k == key: return True
+ else: return False
+
+ def __setitem__(self, key, values):
+ """Set a key's value or values.
+
+ Setting a value overwrites all old ones. The value may be a
+ list of Unicode or UTF-8 strings, or a single Unicode or UTF-8
+ string.
+
+ """
+ if not isinstance(values, list):
+ values = [values]
+ try: del(self[key])
+ except KeyError: pass
+ for value in values:
+ if key in _standard_attribute_names:
+ value = unicode(value)
+ elif not isinstance(value, ASFBaseAttribute):
+ if isinstance(value, basestring):
+ value = ASFUnicodeAttribute(value)
+ elif isinstance(value, bool):
+ value = ASFBoolAttribute(value)
+ elif isinstance(value, int):
+ value = ASFDWordAttribute(value)
+ elif isinstance(value, long):
+ value = ASFQWordAttribute(value)
+ self.append((key, value))
+
+ def keys(self):
+ """Return all keys in the comment."""
+ return self and set(zip(*self)[0])
+
+ def as_dict(self):
+ """Return a copy of the comment data in a real dict."""
+ d = {}
+ for key, value in self:
+ d.setdefault(key, []).append(value)
+ return d
+
+
+class ASFBaseAttribute(object):
+ """Generic attribute."""
+ TYPE = None
+
+ def __init__(self, value=None, data=None, language=None,
+ stream=None, **kwargs):
+ self.language = language
+ self.stream = stream
+ if data:
+ self.value = self.parse(data, **kwargs)
+ else:
+ self.value = value
+
+ def __repr__(self):
+ name = "%s(%r" % (type(self).__name__, self.value)
+ if self.language:
+ name += ", language=%d" % self.language
+ if self.stream:
+ name += ", stream=%d" % self.stream
+ name += ")"
+ return name
+
+ def render(self, name):
+ name = name.encode("utf-16-le") + "\x00\x00"
+ data = self._render()
+ return (struct.pack("<H", len(name)) + name +
+ struct.pack("<HH", self.TYPE, len(data)) + data)
+
+ def render_m(self, name):
+ name = name.encode("utf-16-le") + "\x00\x00"
+ if self.TYPE == 2:
+ data = self._render(dword=False)
+ else:
+ data = self._render()
+ return (struct.pack("<HHHHI", 0, self.stream or 0, len(name),
+ self.TYPE, len(data)) + name + data)
+
+ def render_ml(self, name):
+ name = name.encode("utf-16-le") + "\x00\x00"
+ if self.TYPE == 2:
+ data = self._render(dword=False)
+ else:
+ data = self._render()
+ return (struct.pack("<HHHHI", self.language or 0, self.stream or 0,
+ len(name), self.TYPE, len(data)) + name + data)
+
+class ASFUnicodeAttribute(ASFBaseAttribute):
+ """Unicode string attribute."""
+ TYPE = 0x0000
+
+ def parse(self, data):
+ return data.decode("utf-16-le").strip("\x00")
+
+ def _render(self):
+ return self.value.encode("utf-16-le") + "\x00\x00"
+
+ def __str__(self):
+ return self.value
+
+ def __cmp__(self, other):
+ return cmp(unicode(self), other)
+
+
+class ASFByteArrayAttribute(ASFBaseAttribute):
+ """Byte array attribute."""
+ TYPE = 0x0001
+
+ def parse(self, data):
+ return data
+
+ def _render(self):
+ return self.value
+
+ def __str__(self):
+ return "[binary data (%s bytes)]" % len(self.value)
+
+ def __cmp__(self, other):
+ return cmp(str(self), other)
+
+
+class ASFBoolAttribute(ASFBaseAttribute):
+ """Bool attribute."""
+ TYPE = 0x0002
+
+ def parse(self, data, dword=True):
+ if dword:
+ return struct.unpack("<I", data)[0] == 1
+ else:
+ return struct.unpack("<H", data)[0] == 1
+
+ def _render(self, dword=True):
+ if dword:
+ return struct.pack("<I", int(self.value))
+ else:
+ return struct.pack("<H", int(self.value))
+
+ def __bool__(self):
+ return self.value
+
+ def __str__(self):
+ return str(self.value)
+
+ def __cmp__(self, other):
+ return cmp(bool(self), other)
+
+
+class ASFDWordAttribute(ASFBaseAttribute):
+ """DWORD attribute."""
+ TYPE = 0x0003
+
+ def parse(self, data):
+ return struct.unpack("<L", data)[0]
+
+ def _render(self):
+ return struct.pack("<L", self.value)
+
+ def __int__(self):
+ return self.value
+
+ def __str__(self):
+ return str(self.value)
+
+ def __cmp__(self, other):
+ return cmp(int(self), other)
+
+
+class ASFQWordAttribute(ASFBaseAttribute):
+ """QWORD attribute."""
+ TYPE = 0x0004
+
+ def parse(self, data):
+ return struct.unpack("<Q", data)[0]
+
+ def _render(self):
+ return struct.pack("<Q", self.value)
+
+ def __int__(self):
+ return self.value
+
+ def __str__(self):
+ return str(self.value)
+
+ def __cmp__(self, other):
+ return cmp(int(self), other)
+
+
+class ASFWordAttribute(ASFBaseAttribute):
+ """WORD attribute."""
+ TYPE = 0x0005
+
+ def parse(self, data):
+ return struct.unpack("<H", data)[0]
+
+ def _render(self):
+ return struct.pack("<H", self.value)
+
+ def __int__(self):
+ return self.value
+
+ def __str__(self):
+ return str(self.value)
+
+ def __cmp__(self, other):
+ return cmp(int(self), other)
+
+
+class ASFGUIDAttribute(ASFBaseAttribute):
+ """GUID attribute."""
+ TYPE = 0x0006
+
+ def parse(self, data):
+ return data
+
+ def _render(self):
+ return self.value
+
+ def __str__(self):
+ return self.value
+
+ def __cmp__(self, other):
+ return cmp(str(self), other)
+
+
+UNICODE = ASFUnicodeAttribute.TYPE
+BYTEARRAY = ASFByteArrayAttribute.TYPE
+BOOL = ASFBoolAttribute.TYPE
+DWORD = ASFDWordAttribute.TYPE
+QWORD = ASFQWordAttribute.TYPE
+WORD = ASFWordAttribute.TYPE
+GUID = ASFGUIDAttribute.TYPE
+
+def ASFValue(value, kind, **kwargs):
+ for t, c in _attribute_types.items():
+ if kind == t:
+ return c(value=value, **kwargs)
+ raise ValueError("Unknown value type")
+
+
+_attribute_types = {
+ ASFUnicodeAttribute.TYPE: ASFUnicodeAttribute,
+ ASFByteArrayAttribute.TYPE: ASFByteArrayAttribute,
+ ASFBoolAttribute.TYPE: ASFBoolAttribute,
+ ASFDWordAttribute.TYPE: ASFDWordAttribute,
+ ASFQWordAttribute.TYPE: ASFQWordAttribute,
+ ASFWordAttribute.TYPE: ASFWordAttribute,
+ ASFGUIDAttribute.TYPE: ASFGUIDAttribute,
+}
+
+
+_standard_attribute_names = [
+ "Title",
+ "Author",
+ "Copyright",
+ "Description",
+ "Rating"
+]
+
+
+class BaseObject(object):
+ """Base ASF object."""
+ GUID = None
+
+ def parse(self, asf, data, fileobj, size):
+ self.data = data
+
+ def render(self, asf):
+ data = self.GUID + struct.pack("<Q", len(self.data) + 24) + self.data
+ size = len(data)
+ return data
+
+
+class UnknownObject(BaseObject):
+ """Unknown ASF object."""
+ def __init__(self, guid):
+ self.GUID = guid
+
+
+class HeaderObject(object):
+ """ASF header."""
+ GUID = "\x30\x26\xB2\x75\x8E\x66\xCF\x11\xA6\xD9\x00\xAA\x00\x62\xCE\x6C"
+
+
+class ContentDescriptionObject(BaseObject):
+ """Content description."""
+ GUID = "\x33\x26\xB2\x75\x8E\x66\xCF\x11\xA6\xD9\x00\xAA\x00\x62\xCE\x6C"
+
+ def parse(self, asf, data, fileobj, size):
+ super(ContentDescriptionObject, self).parse(asf, data, fileobj, size)
+ asf.content_description_obj = self
+ lengths = struct.unpack("<HHHHH", data[:10])
+ texts = []
+ pos = 10
+ for length in lengths:
+ end = pos + length
+ texts.append(data[pos:end].decode("utf-16-le").strip("\x00"))
+ pos = end
+ (asf.tags["Title"], asf.tags["Author"], asf.tags["Copyright"],
+ asf.tags["Description"], asf.tags["Rating"]) = texts
+
+ def render(self, asf):
+ def render_text(name):
+ value = asf.tags.get(name, [])
+ if value and value[0]:
+ return value[0].encode("utf-16-le") + "\x00\x00"
+ else:
+ return ""
+ texts = map(render_text, _standard_attribute_names)
+ data = struct.pack("<HHHHH", *map(str.__len__, texts)) + "".join(texts)
+ return self.GUID + struct.pack("<Q", 24 + len(data)) + data
+
+
+class ExtendedContentDescriptionObject(BaseObject):
+ """Extended content description."""
+ GUID = "\x40\xA4\xD0\xD2\x07\xE3\xD2\x11\x97\xF0\x00\xA0\xC9\x5E\xA8\x50"
+
+ def parse(self, asf, data, fileobj, size):
+ super(ExtendedContentDescriptionObject, self).parse(asf, data, fileobj, size)
+ asf.extended_content_description_obj = self
+ num_attributes, = struct.unpack("<H", data[0:2])
+ pos = 2
+ for i in range(num_attributes):
+ name_length, = struct.unpack("<H", data[pos:pos+2])
+ pos += 2
+ name = data[pos:pos+name_length].decode("utf-16-le").strip("\x00")
+ pos += name_length
+ value_type, value_length = struct.unpack("<HH", data[pos:pos+4])
+ pos += 4
+ value = data[pos:pos+value_length]
+ pos += value_length
+ attr = _attribute_types[value_type](data=value)
+ asf.tags.append((name, attr))
+
+ def render(self, asf):
+ attrs = asf.to_extended_content_description.items()
+ data = "".join([attr.render(name) for (name, attr) in attrs])
+ data = struct.pack("<QH", 26 + len(data), len(attrs)) + data
+ return self.GUID + data
+
+
+class FilePropertiesObject(BaseObject):
+ """File properties."""
+ GUID = "\xA1\xDC\xAB\x8C\x47\xA9\xCF\x11\x8E\xE4\x00\xC0\x0C\x20\x53\x65"
+
+ def parse(self, asf, data, fileobj, size):
+ super(FilePropertiesObject, self).parse(asf, data, fileobj, size)
+ length, _, preroll = struct.unpack("<QQQ", data[40:64])
+ asf.info.length = length / 10000000.0 - preroll / 1000.0
+
+
+class StreamPropertiesObject(BaseObject):
+ """Stream properties."""
+ GUID = "\x91\x07\xDC\xB7\xB7\xA9\xCF\x11\x8E\xE6\x00\xC0\x0C\x20\x53\x65"
+
+ def parse(self, asf, data, fileobj, size):
+ super(StreamPropertiesObject, self).parse(asf, data, fileobj, size)
+ channels, sample_rate, bitrate = struct.unpack("<HII", data[56:66])
+ asf.info.channels = channels
+ asf.info.sample_rate = sample_rate
+ asf.info.bitrate = bitrate * 8
+
+
+class HeaderExtensionObject(BaseObject):
+ """Header extension."""
+ GUID = "\xb5\x03\xbf_.\xa9\xcf\x11\x8e\xe3\x00\xc0\x0c Se"
+
+ def parse(self, asf, data, fileobj, size):
+ super(HeaderExtensionObject, self).parse(asf, data, fileobj, size)
+ asf.header_extension_obj = self
+ datasize, = struct.unpack("<I", data[18:22])
+ datapos = 0
+ self.objects = []
+ while datapos < datasize:
+ guid, size = struct.unpack("<16sQ", data[22+datapos:22+datapos+24])
+ if guid in _object_types:
+ obj = _object_types[guid]()
+ else:
+ obj = UnknownObject(guid)
+ obj.parse(asf, data[22+datapos+24:22+datapos+size], fileobj, size)
+ self.objects.append(obj)
+ datapos += size
+
+ def render(self, asf):
+ data = "".join([obj.render(asf) for obj in self.objects])
+ return (self.GUID + struct.pack("<Q", 24 + 16 + 6 + len(data)) +
+ "\x11\xD2\xD3\xAB\xBA\xA9\xcf\x11" +
+ "\x8E\xE6\x00\xC0\x0C\x20\x53\x65" +
+ "\x06\x00" + struct.pack("<I", len(data)) + data)
+
+
+class MetadataObject(BaseObject):
+ """Metadata description."""
+ GUID = "\xea\xcb\xf8\xc5\xaf[wH\x84g\xaa\x8cD\xfaL\xca"
+
+ def parse(self, asf, data, fileobj, size):
+ super(MetadataObject, self).parse(asf, data, fileobj, size)
+ asf.metadata_obj = self
+ num_attributes, = struct.unpack("<H", data[0:2])
+ pos = 2
+ for i in range(num_attributes):
+ (reserved, stream, name_length, value_type,
+ value_length) = struct.unpack("<HHHHI", data[pos:pos+12])
+ pos += 12
+ name = data[pos:pos+name_length].decode("utf-16-le").strip("\x00")
+ pos += name_length
+ value = data[pos:pos+value_length]
+ pos += value_length
+ args = {'data': value, 'stream': stream}
+ if value_type == 2:
+ args['dword'] = False
+ attr = _attribute_types[value_type](**args)
+ asf.tags.append((name, attr))
+
+ def render(self, asf):
+ attrs = asf.to_metadata.items()
+ data = "".join([attr.render_m(name) for (name, attr) in attrs])
+ return (self.GUID + struct.pack("<QH", 26 + len(data), len(attrs)) +
+ data)
+
+
+class MetadataLibraryObject(BaseObject):
+ """Metadata library description."""
+ GUID = "\x94\x1c#D\x98\x94\xd1I\xa1A\x1d\x13NEpT"
+
+ def parse(self, asf, data, fileobj, size):
+ super(MetadataLibraryObject, self).parse(asf, data, fileobj, size)
+ asf.metadata_library_obj = self
+ num_attributes, = struct.unpack("<H", data[0:2])
+ pos = 2
+ for i in range(num_attributes):
+ (language, stream, name_length, value_type,
+ value_length) = struct.unpack("<HHHHI", data[pos:pos+12])
+ pos += 12
+ name = data[pos:pos+name_length].decode("utf-16-le").strip("\x00")
+ pos += name_length
+ value = data[pos:pos+value_length]
+ pos += value_length
+ args = {'data': value, 'language': language, 'stream': stream}
+ if value_type == 2:
+ args['dword'] = False
+ attr = _attribute_types[value_type](**args)
+ asf.tags.append((name, attr))
+
+ def render(self, asf):
+ attrs = asf.to_metadata_library
+ data = "".join([attr.render_ml(name) for (name, attr) in attrs])
+ return (self.GUID + struct.pack("<QH", 26 + len(data), len(attrs)) +
+ data)
+
+
+_object_types = {
+ ExtendedContentDescriptionObject.GUID: ExtendedContentDescriptionObject,
+ ContentDescriptionObject.GUID: ContentDescriptionObject,
+ FilePropertiesObject.GUID: FilePropertiesObject,
+ StreamPropertiesObject.GUID: StreamPropertiesObject,
+ HeaderExtensionObject.GUID: HeaderExtensionObject,
+ MetadataLibraryObject.GUID: MetadataLibraryObject,
+ MetadataObject.GUID: MetadataObject,
+}
+
+
+class ASF(FileType):
+ """An ASF file, probably containing WMA or WMV."""
+
+ _mimes = ["audio/x-ms-wma", "audio/x-ms-wmv", "video/x-ms-asf",
+ "audio/x-wma", "video/x-wmv"]
+
+ def load(self, filename):
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ self.size = 0
+ self.size1 = 0
+ self.size2 = 0
+ self.offset1 = 0
+ self.offset2 = 0
+ self.num_objects = 0
+ self.info = ASFInfo()
+ self.tags = ASFTags()
+ self.__read_file(fileobj)
+ finally:
+ fileobj.close()
+
+ def save(self):
+ # Move attributes to the right objects
+ self.to_extended_content_description = {}
+ self.to_metadata = {}
+ self.to_metadata_library = []
+ for name, value in self.tags:
+ if name in _standard_attribute_names:
+ continue
+ if (value.language is None and value.stream is None and
+ name not in self.to_extended_content_description):
+ self.to_extended_content_description[name] = value
+ elif (value.language is None and value.stream is not None and
+ name not in self.to_metadata):
+ self.to_metadata[name] = value
+ else:
+ self.to_metadata_library.append((name, value))
+
+ # Add missing objects
+ if not self.content_description_obj:
+ self.content_description_obj = \
+ ContentDescriptionObject()
+ self.objects.append(self.content_description_obj)
+ if not self.extended_content_description_obj:
+ self.extended_content_description_obj = \
+ ExtendedContentDescriptionObject()
+ self.objects.append(self.extended_content_description_obj)
+ if not self.header_extension_obj:
+ self.header_extension_obj = \
+ HeaderExtensionObject()
+ self.objects.append(self.header_extension_obj)
+ if not self.metadata_obj:
+ self.metadata_obj = \
+ MetadataObject()
+ self.header_extension_obj.objects.append(self.metadata_obj)
+ if not self.metadata_library_obj:
+ self.metadata_library_obj = \
+ MetadataLibraryObject()
+ self.header_extension_obj.objects.append(self.metadata_library_obj)
+
+ # Render the header
+ data = "".join([obj.render(self) for obj in self.objects])
+ data = (HeaderObject.GUID +
+ struct.pack("<QL", len(data) + 30, len(self.objects)) +
+ "\x01\x02" + data)
+
+ fileobj = file(self.filename, "rb+")
+ try:
+ size = len(data)
+ if size > self.size:
+ insert_bytes(fileobj, size - self.size, self.size)
+ if size < self.size:
+ delete_bytes(fileobj, self.size - size, 0)
+ fileobj.seek(0)
+ fileobj.write(data)
+ finally:
+ fileobj.close()
+
+ def __read_file(self, fileobj):
+ header = fileobj.read(30)
+ if len(header) != 30 or header[:16] != HeaderObject.GUID:
+ raise ASFHeaderError, "Not an ASF file."
+
+ self.extended_content_description_obj = None
+ self.content_description_obj = None
+ self.header_extension_obj = None
+ self.metadata_obj = None
+ self.metadata_library_obj = None
+
+ self.size, self.num_objects = struct.unpack("<QL", header[16:28])
+ self.objects = []
+ for i in range(self.num_objects):
+ self.__read_object(fileobj)
+
+ def __read_object(self, fileobj):
+ guid, size = struct.unpack("<16sQ", fileobj.read(24))
+ if guid in _object_types:
+ obj = _object_types[guid]()
+ else:
+ obj = UnknownObject(guid)
+ data = fileobj.read(size - 24)
+ obj.parse(self, data, fileobj, size)
+ self.objects.append(obj)
+
+ def score(filename, fileobj, header):
+ return header.startswith(HeaderObject.GUID) * 2
+ score = staticmethod(score)
+
+Open = ASF
--- /dev/null
+# Simpler (but far more limited) API for ID3 editing
+# Copyright 2006 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+#
+# $Id: id3.py 3086 2006-04-04 02:13:21Z piman $
+
+"""Easier access to ID3 tags.
+
+EasyID3 is a wrapper around mutagen.id3.ID3 to make ID3 tags appear
+more like Vorbis or APEv2 tags.
+"""
+
+import mutagen.id3
+from mutagen import Metadata
+from mutagen._util import DictMixin
+from mutagen.id3 import ID3, error, delete
+
+__all__ = ['EasyID3', 'Open', 'delete']
+
+class EasyID3(DictMixin, Metadata):
+ """A file with an ID3 tag.
+
+ Like Vorbis comments, EasyID3 keys are case-insensitive ASCII
+ values. Only a subset of ID3 frames (those with simple text keys)
+ are supported; EasyID3.valid_keys maps human-readable EasyID3
+ names to ID3 frame IDs.
+
+ To use an EasyID3 class with mutagen.mp3.MP3:
+ from mutagen.mp3 import MP3
+ from mutagen.easyid3 import EasyID3
+ MP3(filename, ID3=EasyID3)
+ """
+
+ valid_keys = {
+ "album": "TALB",
+ "composer": "TCOM",
+ "genre": "TCON",
+ "date": "TDRC",
+ "lyricist": "TEXT",
+ "title": "TIT2",
+ "version": "TIT3",
+ "artist": "TPE1",
+ "tracknumber": "TRCK",
+ }
+ """Valid keys for EasyID3 instances."""
+
+ def __init__(self, filename=None):
+ self.__id3 = ID3()
+ self.load = self.__id3.load
+ self.save = self.__id3.save
+ self.delete = self.__id3.delete
+ if filename is not None:
+ self.load(filename)
+
+ filename = property(lambda s: s.__id3.filename,
+ lambda s, fn: setattr(s.__id3, 'filename', fn))
+
+ _size = property(lambda s: s._id3.size,
+ lambda s, fn: setattr(s.__id3, '_size', fn))
+
+ def __TCON_get(self, frame):
+ return frame.genres
+
+ def __TCON_set(self, frame, value):
+ frame.encoding = 3
+ if not isinstance(value, list):
+ value = [value]
+ frame.genres = value
+
+ def __TDRC_get(self, frame):
+ return [stamp.text for stamp in frame.text]
+
+ def __TDRC_set(self, frame, value):
+ self.__id3.add(mutagen.id3.TDRC(encoding=3, text=value))
+
+ def __text_get(self, frame):
+ return list(frame)
+
+ def __text_set(self, frame, value):
+ frame.encoding = 3
+ if not isinstance(value, list):
+ value = [value]
+ frame.text = value
+
+ def __getitem__(self, key):
+ key = key.lower()
+ if key in self.valid_keys:
+ frame = self.valid_keys[key]
+ getter = self.__mungers.get(frame, self.__default)[0]
+ return getter(self, self.__id3[frame])
+ else: raise ValueError("%r is not a valid key" % key)
+
+ def __setitem__(self, key, value):
+ key = key.lower()
+ if key in self.valid_keys:
+ frame = self.valid_keys[key]
+ setter = self.__mungers.get(frame, self.__default)[1]
+ if frame not in self.__id3:
+ frame = mutagen.id3.Frames[frame](encoding=3, text=value)
+ self.__id3.loaded_frame(frame)
+ else:
+ setter(self, self.__id3[frame], value)
+ else: raise ValueError("%r is not a valid key" % key)
+
+ def __delitem__(self, key):
+ key = key.lower()
+ if key in self.valid_keys:
+ del(self.__id3[self.valid_keys[key]])
+ else: raise ValueError("%r is not a valid key" % key)
+
+ def keys(self):
+ return [k for (k, v) in self.valid_keys.items() if v in self.__id3]
+
+ def pprint(self):
+ """Print tag key=value pairs."""
+ strings = []
+ for key in self.keys():
+ values = self[key]
+ for value in values:
+ strings.append("%s=%s" % (key, value))
+ return "\n".join(strings)
+
+ __mungers = {
+ "TCON": (__TCON_get, __TCON_set),
+ "TDRC": (__TDRC_get, __TDRC_set),
+ }
+
+ __default = (__text_get, __text_set)
+
+Open = EasyID3
--- /dev/null
+# FLAC comment support for Mutagen
+# Copyright 2005 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+
+"""Read and write FLAC Vorbis comments and stream information.
+
+Read more about FLAC at http://flac.sourceforge.net.
+
+FLAC supports arbitrary metadata blocks. The two most interesting ones
+are the FLAC stream information block, and the Vorbis comment block;
+these are also the only ones Mutagen can currently read.
+
+This module does not handle Ogg FLAC files.
+
+Based off documentation available at
+http://flac.sourceforge.net/format.html
+"""
+
+__all__ = ["FLAC", "Open", "delete"]
+
+import struct
+from cStringIO import StringIO
+from _vorbis import VCommentDict
+from mutagen import FileType
+from mutagen._util import insert_bytes
+from mutagen.id3 import BitPaddedInt
+
+class error(IOError): pass
+class FLACNoHeaderError(error): pass
+class FLACVorbisError(ValueError, error): pass
+
+def to_int_be(string):
+ """Convert an arbitrarily-long string to a long using big-endian
+ byte order."""
+ return reduce(lambda a, b: (a << 8) + ord(b), string, 0L)
+
+class MetadataBlock(object):
+ """A generic block of FLAC metadata.
+
+ This class is extended by specific used as an ancestor for more specific
+ blocks, and also as a container for data blobs of unknown blocks.
+
+ Attributes:
+ data -- raw binary data for this block
+ """
+
+ def __init__(self, data):
+ """Parse the given data string or file-like as a metadata block.
+ The metadata header should not be included."""
+ if data is not None:
+ if isinstance(data, str): data = StringIO(data)
+ elif not hasattr(data, 'read'):
+ raise TypeError(
+ "StreamInfo requires string data or a file-like")
+ self.load(data)
+
+ def load(self, data): self.data = data.read()
+ def write(self): return self.data
+
+ def writeblocks(blocks):
+ """Render metadata block as a byte string."""
+ data = []
+ codes = [[block.code, block.write()] for block in blocks]
+ codes[-1][0] |= 128
+ for code, datum in codes:
+ byte = chr(code)
+ if len(datum) > 2**24:
+ raise error("block is too long to write")
+ length = struct.pack(">I", len(datum))[-3:]
+ data.append(byte + length + datum)
+ return "".join(data)
+ writeblocks = staticmethod(writeblocks)
+
+ def group_padding(blocks):
+ """Consolidate FLAC padding metadata blocks.
+
+ The overall size of the rendered blocks does not change, so
+ this adds several bytes of padding for each merged block."""
+ paddings = filter(lambda x: isinstance(x, Padding), blocks)
+ map(blocks.remove, paddings)
+ padding = Padding()
+ # total padding size is the sum of padding sizes plus 4 bytes
+ # per removed header.
+ size = sum([padding.length for padding in paddings])
+ padding.length = size + 4 * (len(paddings) - 1)
+ blocks.append(padding)
+ group_padding = staticmethod(group_padding)
+
+class StreamInfo(MetadataBlock):
+ """FLAC stream information.
+
+ This contains information about the audio data in the FLAC file.
+ Unlike most stream information objects in Mutagen, changes to this
+ one will rewritten to the file when it is saved. Unless you are
+ actually changing the audio stream itself, don't change any
+ attributes of this block.
+
+ Attributes:
+ min_blocksize -- minimum audio block size
+ max_blocksize -- maximum audio block size
+ sample_rate -- audio sample rate in Hz
+ channels -- audio channels (1 for mono, 2 for stereo)
+ bits_per_sample -- bits per sample
+ total_samples -- total samples in file
+ length -- audio length in seconds
+ """
+
+ code = 0
+
+ def __eq__(self, other):
+ try: return (self.min_blocksize == other.min_blocksize and
+ self.max_blocksize == other.max_blocksize and
+ self.sample_rate == other.sample_rate and
+ self.channels == other.channels and
+ self.bits_per_sample == other.bits_per_sample and
+ self.total_samples == other.total_samples)
+ except: return False
+
+ def load(self, data):
+ self.min_blocksize = int(to_int_be(data.read(2)))
+ self.max_blocksize = int(to_int_be(data.read(2)))
+ self.min_framesize = int(to_int_be(data.read(3)))
+ self.max_framesize = int(to_int_be(data.read(3)))
+ # first 16 bits of sample rate
+ sample_first = to_int_be(data.read(2))
+ # last 4 bits of sample rate, 3 of channels, first 1 of bits/sample
+ sample_channels_bps = to_int_be(data.read(1))
+ # last 4 of bits/sample, 36 of total samples
+ bps_total = to_int_be(data.read(5))
+
+ sample_tail = sample_channels_bps >> 4
+ self.sample_rate = int((sample_first << 4) + sample_tail)
+ self.channels = int(((sample_channels_bps >> 1) & 7) + 1)
+ bps_tail = bps_total >> 36
+ bps_head = (sample_channels_bps & 1) << 4
+ self.bits_per_sample = int(bps_head + bps_tail + 1)
+ self.total_samples = bps_total & 0xFFFFFFFFFL
+ self.length = self.total_samples / float(self.sample_rate)
+
+ self.md5_signature = to_int_be(data.read(16))
+
+ def write(self):
+ f = StringIO()
+ f.write(struct.pack(">I", self.min_blocksize)[-2:])
+ f.write(struct.pack(">I", self.max_blocksize)[-2:])
+ f.write(struct.pack(">I", self.min_framesize)[-3:])
+ f.write(struct.pack(">I", self.max_framesize)[-3:])
+
+ # first 16 bits of sample rate
+ f.write(struct.pack(">I", self.sample_rate >> 4)[-2:])
+ # 4 bits sample, 3 channel, 1 bps
+ byte = (self.sample_rate & 0xF) << 4
+ byte += ((self.channels - 1) & 3) << 1
+ byte += ((self.bits_per_sample - 1) >> 4) & 1
+ f.write(chr(byte))
+ # 4 bits of bps, 4 of sample count
+ byte = ((self.bits_per_sample - 1) & 0xF) << 4
+ byte += (self.total_samples >> 32) & 0xF
+ f.write(chr(byte))
+ # last 32 of sample count
+ f.write(struct.pack(">I", self.total_samples & 0xFFFFFFFFL))
+ # MD5 signature
+ sig = self.md5_signature
+ f.write(struct.pack(
+ ">4I", (sig >> 96) & 0xFFFFFFFFL, (sig >> 64) & 0xFFFFFFFFL,
+ (sig >> 32) & 0xFFFFFFFFL, sig & 0xFFFFFFFFL))
+ return f.getvalue()
+
+ def pprint(self):
+ return "FLAC, %.2f seconds, %d Hz" % (self.length, self.sample_rate)
+
+class SeekPoint(tuple):
+ """A single seek point in a FLAC file.
+
+ Placeholder seek points have first_sample of 0xFFFFFFFFFFFFFFFFL,
+ and byte_offset and num_samples undefined. Seek points must be
+ sorted in ascending order by first_sample number. Seek points must
+ be unique by first_sample number, except for placeholder
+ points. Placeholder points must occur last in the table and there
+ may be any number of them.
+
+ Attributes:
+ first_sample -- sample number of first sample in the target frame
+ byte_offset -- offset from first frame to target frame
+ num_samples -- number of samples in target frame
+ """
+
+ def __new__(cls, first_sample, byte_offset, num_samples):
+ return super(cls, SeekPoint).__new__(cls, (first_sample,
+ byte_offset, num_samples))
+ first_sample = property(lambda self: self[0])
+ byte_offset = property(lambda self: self[1])
+ num_samples = property(lambda self: self[2])
+
+class SeekTable(MetadataBlock):
+ """Read and write FLAC seek tables.
+
+ Attributes:
+ seekpoints -- list of SeekPoint objects
+ """
+
+ __SEEKPOINT_FORMAT = '>QQH'
+ __SEEKPOINT_SIZE = struct.calcsize(__SEEKPOINT_FORMAT)
+
+ code = 3
+
+ def __init__(self, data):
+ self.seekpoints = []
+ super(SeekTable, self).__init__(data)
+
+ def __eq__(self, other):
+ try: return (self.seekpoints == other.seekpoints)
+ except (AttributeError, TypeError): return False
+
+ def load(self, data):
+ self.seekpoints = []
+ sp = data.read(self.__SEEKPOINT_SIZE)
+ while len(sp) == self.__SEEKPOINT_SIZE:
+ self.seekpoints.append(SeekPoint(
+ *struct.unpack(self.__SEEKPOINT_FORMAT, sp)))
+ sp = data.read(self.__SEEKPOINT_SIZE)
+
+ def write(self):
+ f = StringIO()
+ for seekpoint in self.seekpoints:
+ packed = struct.pack(self.__SEEKPOINT_FORMAT,
+ seekpoint.first_sample, seekpoint.byte_offset,
+ seekpoint.num_samples)
+ f.write(packed)
+ return f.getvalue()
+
+ def __repr__(self):
+ return "<%s seekpoints=%r>" % (type(self).__name__, self.seekpoints)
+
+class VCFLACDict(VCommentDict):
+ """Read and write FLAC Vorbis comments.
+
+ FLACs don't use the framing bit at the end of the comment block.
+ So this extends VCommentDict to not use the framing bit.
+ """
+
+ code = 4
+
+ def load(self, data, errors='replace', framing=False):
+ super(VCFLACDict, self).load(data, errors=errors, framing=framing)
+
+ def write(self, framing=False):
+ return super(VCFLACDict, self).write(framing=framing)
+
+class CueSheetTrackIndex(tuple):
+ """Index for a track in a cuesheet.
+
+ For CD-DA, an index_number of 0 corresponds to the track
+ pre-gap. The first index in a track must have a number of 0 or 1,
+ and subsequently, index_numbers must increase by 1. Index_numbers
+ must be unique within a track. And index_offset must be evenly
+ divisible by 588 samples.
+
+ Attributes:
+ index_number -- index point number
+ index_offset -- offset in samples from track start
+ """
+
+ def __new__(cls, index_number, index_offset):
+ return super(cls, CueSheetTrackIndex).__new__(cls,
+ (index_number, index_offset))
+ index_number = property(lambda self: self[0])
+ index_offset = property(lambda self: self[1])
+
+class CueSheetTrack(object):
+ """A track in a cuesheet.
+
+ For CD-DA, track_numbers must be 1-99, or 170 for the
+ lead-out. Track_numbers must be unique within a cue sheet. There
+ must be atleast one index in every track except the lead-out track
+ which must have none.
+
+ Attributes:
+ track_number -- track number
+ start_offset -- track offset in samples from start of FLAC stream
+ isrc -- ISRC code
+ type -- 0 for audio, 1 for digital data
+ pre_emphasis -- true if the track is recorded with pre-emphasis
+ indexes -- list of CueSheetTrackIndex objects
+ """
+
+ def __init__(self, track_number, start_offset, isrc='', type_=0,
+ pre_emphasis=False):
+ self.track_number = track_number
+ self.start_offset = start_offset
+ self.isrc = isrc
+ self.type = type_
+ self.pre_emphasis = pre_emphasis
+ self.indexes = []
+
+ def __eq__(self, other):
+ try: return (self.track_number == other.track_number and
+ self.start_offset == other.start_offset and
+ self.isrc == other.isrc and
+ self.type == other.type and
+ self.pre_emphasis == other.pre_emphasis and
+ self.indexes == other.indexes)
+ except (AttributeError, TypeError): return False
+
+ def __repr__(self):
+ return ("<%s number=%r, offset=%d, isrc=%r, type=%r, "
+ "pre_emphasis=%r, indexes=%r)>") % (
+ type(self).__name__, self.track_number, self.start_offset,
+ self.isrc, self.type, self.pre_emphasis, self.indexes)
+
+class CueSheet(MetadataBlock):
+ """Read and write FLAC embedded cue sheets.
+
+ Number of tracks should be from 1 to 100. There should always be
+ exactly one lead-out track and that track must be the last track
+ in the cue sheet.
+
+ Attributes:
+ media_catalog_number -- media catalog number in ASCII
+ lead_in_samples -- number of lead-in samples
+ compact_disc -- true if the cuesheet corresponds to a compact disc
+ tracks -- list of CueSheetTrack objects
+ lead_out -- lead-out as CueSheetTrack or None if lead-out was not found
+ """
+
+ __CUESHEET_FORMAT = '>128sQB258xB'
+ __CUESHEET_SIZE = struct.calcsize(__CUESHEET_FORMAT)
+ __CUESHEET_TRACK_FORMAT = '>QB12sB13xB'
+ __CUESHEET_TRACK_SIZE = struct.calcsize(__CUESHEET_TRACK_FORMAT)
+ __CUESHEET_TRACKINDEX_FORMAT = '>QB3x'
+ __CUESHEET_TRACKINDEX_SIZE = struct.calcsize(__CUESHEET_TRACKINDEX_FORMAT)
+
+ code = 5
+
+ media_catalog_number = ''
+ lead_in_samples = 88200
+ compact_disc = True
+
+ def __init__(self, data):
+ self.tracks = []
+ super(CueSheet, self).__init__(data)
+
+ def __eq__(self, other):
+ try:
+ return (self.media_catalog_number == other.media_catalog_number and
+ self.lead_in_samples == other.lead_in_samples and
+ self.compact_disc == other.compact_disc and
+ self.tracks == other.tracks)
+ except (AttributeError, TypeError): return False
+
+ def load(self, data):
+ header = data.read(self.__CUESHEET_SIZE)
+ media_catalog_number, lead_in_samples, flags, num_tracks = \
+ struct.unpack(self.__CUESHEET_FORMAT, header)
+ self.media_catalog_number = media_catalog_number.rstrip('\0')
+ self.lead_in_samples = lead_in_samples
+ self.compact_disc = bool(flags & 0x80)
+ self.tracks = []
+ for i in range(num_tracks):
+ track = data.read(self.__CUESHEET_TRACK_SIZE)
+ start_offset, track_number, isrc_padded, flags, num_indexes = \
+ struct.unpack(self.__CUESHEET_TRACK_FORMAT, track)
+ isrc = isrc_padded.rstrip('\0')
+ type_ = (flags & 0x80) >> 7
+ pre_emphasis = bool(flags & 0x40)
+ val = CueSheetTrack(
+ track_number, start_offset, isrc, type_, pre_emphasis)
+ for j in range(num_indexes):
+ index = data.read(self.__CUESHEET_TRACKINDEX_SIZE)
+ index_offset, index_number = struct.unpack(
+ self.__CUESHEET_TRACKINDEX_FORMAT, index)
+ val.indexes.append(
+ CueSheetTrackIndex(index_number, index_offset))
+ self.tracks.append(val)
+
+ def write(self):
+ f = StringIO()
+ flags = 0
+ if self.compact_disc: flags |= 0x80
+ packed = struct.pack(
+ self.__CUESHEET_FORMAT, self.media_catalog_number,
+ self.lead_in_samples, flags, len(self.tracks))
+ f.write(packed)
+ for track in self.tracks:
+ track_flags = 0
+ track_flags |= (track.type & 1) << 7
+ if track.pre_emphasis: track_flags |= 0x40
+ track_packed = struct.pack(
+ self.__CUESHEET_TRACK_FORMAT, track.start_offset,
+ track.track_number, track.isrc, track_flags,
+ len(track.indexes))
+ f.write(track_packed)
+ for index in track.indexes:
+ index_packed = struct.pack(
+ self.__CUESHEET_TRACKINDEX_FORMAT,
+ index.index_offset, index.index_number)
+ f.write(index_packed)
+ return f.getvalue()
+
+ def __repr__(self):
+ return ("<%s media_catalog_number=%r, lead_in=%r, compact_disc=%r, "
+ "tracks=%r>") % (
+ type(self).__name__, self.media_catalog_number,
+ self.lead_in_samples, self.compact_disc, self.tracks)
+
+class Picture(MetadataBlock):
+ """Read and write FLAC embed pictures.
+
+ Attributes:
+ type -- picture type (same as types for ID3 APIC frames)
+ mime -- MIME type of the picture
+ desc -- picture's description
+ width -- width in pixels
+ height -- height in pixels
+ depth -- color depth in bits-per-pixel
+ colors -- number of colors for indexed palettes (like GIF),
+ 0 for non-indexed
+ data -- picture data
+ """
+
+ code = 6
+
+ def __init__(self, data=None):
+ self.type = 0
+ self.mime = u''
+ self.desc = u''
+ self.width = 0
+ self.height = 0
+ self.depth = 0
+ self.colors = 0
+ self.data = ''
+ super(Picture, self).__init__(data)
+
+ def __eq__(self, other):
+ try: return (self.type == other.type and
+ self.mime == other.mime and
+ self.desc == other.desc and
+ self.width == other.width and
+ self.height == other.height and
+ self.depth == other.depth and
+ self.colors == other.colors and
+ self.data == other.data)
+ except (AttributeError, TypeError): return False
+
+ def load(self, data):
+ self.type, length = struct.unpack('>2I', data.read(8))
+ self.mime = data.read(length).decode('UTF-8', 'replace')
+ length, = struct.unpack('>I', data.read(4))
+ self.desc = data.read(length).decode('UTF-8', 'replace')
+ (self.width, self.height, self.depth,
+ self.colors, length) = struct.unpack('>5I', data.read(20))
+ self.data = data.read(length)
+
+ def write(self):
+ f = StringIO()
+ mime = self.mime.encode('UTF-8')
+ f.write(struct.pack('>2I', self.type, len(mime)))
+ f.write(mime)
+ desc = self.desc.encode('UTF-8')
+ f.write(struct.pack('>I', len(desc)))
+ f.write(desc)
+ f.write(struct.pack('>5I', self.width, self.height, self.depth,
+ self.colors, len(self.data)))
+ f.write(self.data)
+ return f.getvalue()
+
+ def __repr__(self):
+ return "<%s '%s' (%d bytes)>" % (type(self).__name__, self.mime,
+ len(self.data))
+
+class Padding(MetadataBlock):
+ """Empty padding space for metadata blocks.
+
+ To avoid rewriting the entire FLAC file when editing comments,
+ metadata is often padded. Padding should occur at the end, and no
+ more than one padding block should be in any FLAC file. Mutagen
+ handles this with MetadataBlock.group_padding.
+ """
+
+ code = 1
+
+ def __init__(self, data=""): super(Padding, self).__init__(data)
+ def load(self, data): self.length = len(data.read())
+ def write(self):
+ try: return "\x00" * self.length
+ # On some 64 bit platforms this won't generate a MemoryError
+ # or OverflowError since you might have enough RAM, but it
+ # still generates a ValueError. On other 64 bit platforms,
+ # this will still succeed for extremely large values.
+ # Those should never happen in the real world, and if they
+ # do, writeblocks will catch it.
+ except (OverflowError, ValueError, MemoryError):
+ raise error("cannot write %d bytes" % self.length)
+ def __eq__(self, other):
+ return isinstance(other, Padding) and self.length == other.length
+ def __repr__(self):
+ return "<%s (%d bytes)>" % (type(self).__name__, self.length)
+
+class FLAC(FileType):
+ """A FLAC audio file.
+
+ Attributes:
+ info -- stream information (length, bitrate, sample rate)
+ tags -- metadata tags, if any
+ cuesheet -- CueSheet object, if any
+ seektable -- SeekTable object, if any
+ pictures -- list of embedded pictures
+ """
+
+ _mimes = ["audio/x-flac", "application/x-flac"]
+
+ METADATA_BLOCKS = [StreamInfo, Padding, None, SeekTable, VCFLACDict,
+ CueSheet, Picture]
+ """Known metadata block types, indexed by ID."""
+
+ def score(filename, fileobj, header):
+ return header.startswith("fLaC")
+ score = staticmethod(score)
+
+ def __read_metadata_block(self, file):
+ byte = ord(file.read(1))
+ size = to_int_be(file.read(3))
+ try:
+ data = file.read(size)
+ if len(data) != size:
+ raise error(
+ "file said %d bytes, read %d bytes" % (size, len(data)))
+ block = self.METADATA_BLOCKS[byte & 0x7F](data)
+ except (IndexError, TypeError):
+ block = MetadataBlock(data)
+ block.code = byte & 0x7F
+ self.metadata_blocks.append(block)
+ else:
+ self.metadata_blocks.append(block)
+ if block.code == VCFLACDict.code:
+ if self.tags is None: self.tags = block
+ else: raise FLACVorbisError("> 1 Vorbis comment block found")
+ elif block.code == CueSheet.code:
+ if self.cuesheet is None: self.cuesheet = block
+ else: raise error("> 1 CueSheet block found")
+ elif block.code == SeekTable.code:
+ if self.seektable is None: self.seektable = block
+ else: raise error("> 1 SeekTable block found")
+ return (byte >> 7) ^ 1
+
+ def add_tags(self):
+ """Add a Vorbis comment block to the file."""
+ if self.tags is None:
+ self.tags = VCFLACDict()
+ self.metadata_blocks.append(self.tags)
+ else: raise FLACVorbisError("a Vorbis comment already exists")
+ add_vorbiscomment = add_tags
+
+ def delete(self, filename=None):
+ """Remove Vorbis comments from a file.
+
+ If no filename is given, the one most recently loaded is used.
+ """
+ if filename is None: filename = self.filename
+ for s in list(self.metadata_blocks):
+ if isinstance(s, VCFLACDict):
+ self.metadata_blocks.remove(s)
+ self.tags = None
+ self.save()
+ break
+
+ vc = property(lambda s: s.tags, doc="Alias for tags; don't use this.")
+
+ def load(self, filename):
+ """Load file information from a filename."""
+
+ self.metadata_blocks = []
+ self.tags = None
+ self.cuesheet = None
+ self.seektable = None
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ self.__check_header(fileobj)
+ while self.__read_metadata_block(fileobj): pass
+ finally:
+ fileobj.close()
+
+ try: self.metadata_blocks[0].length
+ except (AttributeError, IndexError):
+ raise FLACNoHeaderError("Stream info block not found")
+
+ info = property(lambda s: s.metadata_blocks[0])
+
+ def add_picture(self, picture):
+ """Add a new picture to the file."""
+ self.metadata_blocks.append(picture)
+
+ def clear_pictures(self):
+ """Delete all pictures from the file."""
+ self.metadata_blocks = filter(lambda b: b.code != Picture.code,
+ self.metadata_blocks)
+
+ def __get_pictures(self):
+ return filter(lambda b: b.code == Picture.code, self.metadata_blocks)
+ pictures = property(__get_pictures, doc="List of embedded pictures")
+
+ def save(self, filename=None, deleteid3=False):
+ """Save metadata blocks to a file.
+
+ If no filename is given, the one most recently loaded is used.
+ """
+
+ if filename is None: filename = self.filename
+ f = open(filename, 'rb+')
+
+ # Ensure we've got padding at the end, and only at the end.
+ # If adding makes it too large, we'll scale it down later.
+ self.metadata_blocks.append(Padding('\x00' * 1020))
+ MetadataBlock.group_padding(self.metadata_blocks)
+
+ header = self.__check_header(f)
+ available = self.__find_audio_offset(f) - header # "fLaC" and maybe ID3
+ data = MetadataBlock.writeblocks(self.metadata_blocks)
+
+ # Delete ID3v2
+ if deleteid3 and header > 4:
+ available += header - 4
+ header = 4
+
+ if len(data) > available:
+ # If we have too much data, see if we can reduce padding.
+ padding = self.metadata_blocks[-1]
+ newlength = padding.length - (len(data) - available)
+ if newlength > 0:
+ padding.length = newlength
+ data = MetadataBlock.writeblocks(self.metadata_blocks)
+ assert len(data) == available
+
+ elif len(data) < available:
+ # If we have too little data, increase padding.
+ self.metadata_blocks[-1].length += (available - len(data))
+ data = MetadataBlock.writeblocks(self.metadata_blocks)
+ assert len(data) == available
+
+ if len(data) != available:
+ # We couldn't reduce the padding enough.
+ diff = (len(data) - available)
+ insert_bytes(f, diff, header)
+
+ f.seek(header - 4)
+ f.write("fLaC" + data)
+
+ # Delete ID3v1
+ if deleteid3:
+ try: f.seek(-128, 2)
+ except IOError: pass
+ else:
+ if f.read(3) == "TAG":
+ f.seek(-128, 2)
+ f.truncate()
+
+ def __find_audio_offset(self, fileobj):
+ byte = 0x00
+ while not (byte >> 7) & 1:
+ byte = ord(fileobj.read(1))
+ size = to_int_be(fileobj.read(3))
+ fileobj.read(size)
+ return fileobj.tell()
+
+ def __check_header(self, fileobj):
+ size = 4
+ header = fileobj.read(4)
+ if header != "fLaC":
+ size = None
+ if header[:3] == "ID3":
+ size = 14 + BitPaddedInt(fileobj.read(6)[2:])
+ fileobj.seek(size - 4)
+ if fileobj.read(4) != "fLaC": size = None
+ if size is None:
+ raise FLACNoHeaderError(
+ "%r is not a valid FLAC file" % fileobj.name)
+ return size
+
+Open = FLAC
+
+def delete(filename):
+ """Remove tags from a file."""
+ FLAC(filename).delete()
--- /dev/null
+# id3 support for mutagen
+# Copyright (C) 2005 Michael Urman
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+#
+# $Id: id3.py 4275 2008-06-01 06:32:37Z piman $
+
+"""ID3v2 reading and writing.
+
+This is based off of the following references:
+ http://www.id3.org/id3v2.4.0-structure.txt
+ http://www.id3.org/id3v2.4.0-frames.txt
+ http://www.id3.org/id3v2.3.0.html
+ http://www.id3.org/id3v2-00.txt
+ http://www.id3.org/id3v1.html
+
+Its largest deviation from the above (versions 2.3 and 2.2) is that it
+will not interpret the / characters as a separator, and will almost
+always accept null separators to generate multi-valued text frames.
+
+Because ID3 frame structure differs between frame types, each frame is
+implemented as a different class (e.g. TIT2 as mutagen.id3.TIT2). Each
+frame's documentation contains a list of its attributes.
+
+Since this file's documentation is a little unwieldy, you are probably
+interested in the 'ID3' class to start with.
+"""
+
+__all__ = ['ID3', 'ID3FileType', 'Frames', 'Open', 'delete']
+
+import struct; from struct import unpack, pack
+from zlib import error as zlibError
+from warnings import warn
+
+import mutagen
+from mutagen._util import insert_bytes, delete_bytes, DictProxy
+
+class error(Exception): pass
+class ID3NoHeaderError(error, ValueError): pass
+class ID3BadUnsynchData(error, ValueError): pass
+class ID3BadCompressedData(error, ValueError): pass
+class ID3TagError(error, ValueError): pass
+class ID3UnsupportedVersionError(error, NotImplementedError): pass
+class ID3EncryptionUnsupportedError(error, NotImplementedError): pass
+class ID3JunkFrameError(error, ValueError): pass
+
+class ID3Warning(error, UserWarning): pass
+
+def is_valid_frame_id(frame_id):
+ return frame_id.isalnum() and frame_id.isupper()
+
+class ID3(DictProxy, mutagen.Metadata):
+ """A file with an ID3v2 tag.
+
+ Attributes:
+ version -- ID3 tag version as a tuple
+ unknown_frames -- raw frame data of any unknown frames found
+ size -- the total size of the ID3 tag, including the header
+ """
+
+ PEDANTIC = True
+ version = (2, 4, 0)
+
+ filename = None
+ size = 0
+ __flags = 0
+ __readbytes = 0
+ __crc = None
+
+ def __init__(self, *args, **kwargs):
+ self.unknown_frames = []
+ super(ID3, self).__init__(*args, **kwargs)
+
+ def __fullread(self, size):
+ try:
+ if size < 0:
+ raise ValueError('Requested bytes (%s) less than zero' % size)
+ if size > self.__filesize:
+ raise EOFError('Requested %#x of %#x (%s)' %
+ (long(size), long(self.__filesize), self.filename))
+ except AttributeError: pass
+ data = self.__fileobj.read(size)
+ if len(data) != size: raise EOFError
+ self.__readbytes += size
+ return data
+
+ def load(self, filename, known_frames=None, translate=True):
+ """Load tags from a filename.
+
+ Keyword arguments:
+ filename -- filename to load tag data from
+ known_frames -- dict mapping frame IDs to Frame objects
+ translate -- Update all tags to ID3v2.4 internally. Mutagen is
+ only capable of writing ID3v2.4 tags, so if you
+ intend to save, this must be true.
+
+ Example of loading a custom frame:
+ my_frames = dict(mutagen.id3.Frames)
+ class XMYF(Frame): ...
+ my_frames["XMYF"] = XMYF
+ mutagen.id3.ID3(filename, known_frames=my_frames)
+ """
+
+ from os.path import getsize
+ self.filename = filename
+ self.__known_frames = known_frames
+ self.__fileobj = file(filename, 'rb')
+ self.__filesize = getsize(filename)
+ try:
+ try:
+ self.__load_header()
+ except EOFError:
+ self.size = 0
+ raise ID3NoHeaderError("%s: too small (%d bytes)" %(
+ filename, self.__filesize))
+ except (ID3NoHeaderError, ID3UnsupportedVersionError), err:
+ self.size = 0
+ import sys
+ stack = sys.exc_info()[2]
+ try: self.__fileobj.seek(-128, 2)
+ except EnvironmentError: raise err, None, stack
+ else:
+ frames = ParseID3v1(self.__fileobj.read(128))
+ if frames is not None:
+ self.version = (1, 1)
+ map(self.add, frames.values())
+ else: raise err, None, stack
+ else:
+ frames = self.__known_frames
+ if frames is None:
+ if (2,3,0) <= self.version: frames = Frames
+ elif (2,2,0) <= self.version: frames = Frames_2_2
+ data = self.__fullread(self.size - 10)
+ for frame in self.__read_frames(data, frames=frames):
+ if isinstance(frame, Frame): self.add(frame)
+ else: self.unknown_frames.append(frame)
+ finally:
+ self.__fileobj.close()
+ del self.__fileobj
+ del self.__filesize
+ if translate:
+ self.update_to_v24()
+
+ def getall(self, key):
+ """Return all frames with a given name (the list may be empty).
+
+ This is best explained by examples:
+ id3.getall('TIT2') == [id3['TIT2']]
+ id3.getall('TTTT') == []
+ id3.getall('TXXX') == [TXXX(desc='woo', text='bar'),
+ TXXX(desc='baz', text='quuuux'), ...]
+
+ Since this is based on the frame's HashKey, which is
+ colon-separated, you can use it to do things like
+ getall('COMM:MusicMatch') or getall('TXXX:QuodLibet:').
+ """
+ if key in self: return [self[key]]
+ else:
+ key = key + ":"
+ return [v for s,v in self.items() if s.startswith(key)]
+
+ def delall(self, key):
+ """Delete all tags of a given kind; see getall."""
+ if key in self: del(self[key])
+ else:
+ key = key + ":"
+ for k in filter(lambda s: s.startswith(key), self.keys()):
+ del(self[k])
+
+ def setall(self, key, values):
+ """Delete frames of the given type and add frames in 'values'."""
+ self.delall(key)
+ for tag in values:
+ self[tag.HashKey] = tag
+
+ def pprint(self):
+ """Return tags in a human-readable format.
+
+ "Human-readable" is used loosely here. The format is intended
+ to mirror that used for Vorbis or APEv2 output, e.g.
+ TIT2=My Title
+ However, ID3 frames can have multiple keys:
+ POPM=user@example.org=3 128/255
+ """
+ return "\n".join(map(Frame.pprint, self.values()))
+
+ def loaded_frame(self, tag):
+ """Deprecated; use the add method."""
+ # turn 2.2 into 2.3/2.4 tags
+ if len(type(tag).__name__) == 3: tag = type(tag).__base__(tag)
+ self[tag.HashKey] = tag
+
+ # add = loaded_frame (and vice versa) break applications that
+ # expect to be able to override loaded_frame (e.g. Quod Libet),
+ # as does making loaded_frame call add.
+ def add(self, frame):
+ """Add a frame to the tag."""
+ return self.loaded_frame(frame)
+
+ def __load_header(self):
+ fn = self.filename
+ data = self.__fullread(10)
+ id3, vmaj, vrev, flags, size = unpack('>3sBBB4s', data)
+ self.__flags = flags
+ self.size = BitPaddedInt(size) + 10
+ self.version = (2, vmaj, vrev)
+
+ if id3 != 'ID3':
+ raise ID3NoHeaderError("'%s' doesn't start with an ID3 tag" % fn)
+ if vmaj not in [2, 3, 4]:
+ raise ID3UnsupportedVersionError("'%s' ID3v2.%d not supported"
+ % (fn, vmaj))
+
+ if self.PEDANTIC:
+ if (2,4,0) <= self.version and (flags & 0x0f):
+ raise ValueError("'%s' has invalid flags %#02x" % (fn, flags))
+ elif (2,3,0) <= self.version and (flags & 0x1f):
+ raise ValueError("'%s' has invalid flags %#02x" % (fn, flags))
+
+ if self.f_extended:
+ if self.version >= (2,4,0):
+ # "Where the 'Extended header size' is the size of the whole
+ # extended header, stored as a 32 bit synchsafe integer."
+ self.__extsize = BitPaddedInt(self.__fullread(4)) - 4
+ else:
+ # "Where the 'Extended header size', currently 6 or 10 bytes,
+ # excludes itself."
+ self.__extsize = unpack('>L', self.__fullread(4))[0]
+ self.__extdata = self.__fullread(self.__extsize)
+
+ def __determine_bpi(self, data, frames):
+ if self.version < (2,4,0): return int
+ # have to special case whether to use bitpaddedints here
+ # spec says to use them, but iTunes has it wrong
+
+ # count number of tags found as BitPaddedInt and how far past
+ o = 0
+ asbpi = 0
+ while o < len(data)-10:
+ name, size, flags = unpack('>4sLH', data[o:o+10])
+ size = BitPaddedInt(size)
+ o += 10+size
+ if name in frames: asbpi += 1
+ bpioff = o - len(data)
+
+ # count number of tags found as int and how far past
+ o = 0
+ asint = 0
+ while o < len(data)-10:
+ name, size, flags = unpack('>4sLH', data[o:o+10])
+ o += 10+size
+ if name in frames: asint += 1
+ intoff = o - len(data)
+
+ # if more tags as int, or equal and bpi is past and int is not
+ if asint > asbpi or (asint == asbpi and (bpioff >= 1 and intoff <= 1)):
+ return int
+ return BitPaddedInt
+
+ def __read_frames(self, data, frames):
+ if self.version < (2,4,0) and self.f_unsynch:
+ try: data = unsynch.decode(data)
+ except ValueError: pass
+
+ if (2,3,0) <= self.version:
+ bpi = self.__determine_bpi(data, frames)
+ while data:
+ header = data[:10]
+ try: name, size, flags = unpack('>4sLH', header)
+ except struct.error: return # not enough header
+ if name.strip('\x00') == '': return
+ size = bpi(size)
+ framedata = data[10:10+size]
+ data = data[10+size:]
+ if size == 0: continue # drop empty frames
+ try: tag = frames[name]
+ except KeyError:
+ if is_valid_frame_id(name): yield header + framedata
+ else:
+ try: yield self.__load_framedata(tag, flags, framedata)
+ except NotImplementedError: yield header + framedata
+ except ID3JunkFrameError: pass
+
+ elif (2,2,0) <= self.version:
+ while data:
+ header = data[0:6]
+ try: name, size = unpack('>3s3s', header)
+ except struct.error: return # not enough header
+ size, = struct.unpack('>L', '\x00'+size)
+ if name.strip('\x00') == '': return
+ framedata = data[6:6+size]
+ data = data[6+size:]
+ if size == 0: continue # drop empty frames
+ try: tag = frames[name]
+ except KeyError:
+ if is_valid_frame_id(name): yield header + framedata
+ else:
+ try: yield self.__load_framedata(tag, 0, framedata)
+ except NotImplementedError: yield header + framedata
+ except ID3JunkFrameError: pass
+
+ def __load_framedata(self, tag, flags, framedata):
+ return tag.fromData(self, flags, framedata)
+
+ f_unsynch = property(lambda s: bool(s.__flags & 0x80))
+ f_extended = property(lambda s: bool(s.__flags & 0x40))
+ f_experimental = property(lambda s: bool(s.__flags & 0x20))
+ f_footer = property(lambda s: bool(s.__flags & 0x10))
+
+ #f_crc = property(lambda s: bool(s.__extflags & 0x8000))
+
+ def save(self, filename=None, v1=1):
+ """Save changes to a file.
+
+ If no filename is given, the one most recently loaded is used.
+
+ Keyword arguments:
+ v1 -- if 0, ID3v1 tags will be removed
+ if 1, ID3v1 tags will be updated but not added
+ if 2, ID3v1 tags will be created and/or updated
+
+ The lack of a way to update only an ID3v1 tag is intentional.
+ """
+
+ # Sort frames by 'importance'
+ order = ["TIT2", "TPE1", "TRCK", "TALB", "TPOS", "TDRC", "TCON"]
+ order = dict(zip(order, range(len(order))))
+ last = len(order)
+ frames = self.items()
+ frames.sort(lambda a, b: cmp(order.get(a[0][:4], last),
+ order.get(b[0][:4], last)))
+
+ framedata = [self.__save_frame(frame) for (key, frame) in frames]
+ framedata.extend([data for data in self.unknown_frames
+ if len(data) > 10])
+ if not framedata:
+ try:
+ self.delete(filename)
+ except EnvironmentError, err:
+ from errno import ENOENT
+ if err.errno != ENOENT: raise
+ return
+
+ framedata = ''.join(framedata)
+ framesize = len(framedata)
+
+ if filename is None: filename = self.filename
+ try: f = open(filename, 'rb+')
+ except IOError, err:
+ from errno import ENOENT
+ if err.errno != ENOENT: raise
+ f = open(filename, 'ab') # create, then reopen
+ f = open(filename, 'rb+')
+ try:
+ idata = f.read(10)
+ try: id3, vmaj, vrev, flags, insize = unpack('>3sBBB4s', idata)
+ except struct.error: id3, insize = '', 0
+ insize = BitPaddedInt(insize)
+ if id3 != 'ID3': insize = -10
+
+ if insize >= framesize: outsize = insize
+ else: outsize = (framesize + 1023) & ~0x3FF
+ framedata += '\x00' * (outsize - framesize)
+
+ framesize = BitPaddedInt.to_str(outsize, width=4)
+ flags = 0
+ header = pack('>3sBBB4s', 'ID3', 4, 0, flags, framesize)
+ data = header + framedata
+
+ if (insize < outsize):
+ insert_bytes(f, outsize-insize, insize+10)
+ f.seek(0)
+ f.write(data)
+
+ try:
+ f.seek(-128, 2)
+ except IOError, err:
+ from errno import EINVAL
+ if err.errno != EINVAL: raise
+ f.seek(0, 2) # ensure read won't get "TAG"
+
+ if f.read(3) == "TAG":
+ f.seek(-128, 2)
+ if v1 > 0: f.write(MakeID3v1(self))
+ else: f.truncate()
+ elif v1 == 2:
+ f.seek(0, 2)
+ f.write(MakeID3v1(self))
+
+ finally:
+ f.close()
+
+ def delete(self, filename=None, delete_v1=True, delete_v2=True):
+ """Remove tags from a file.
+
+ If no filename is given, the one most recently loaded is used.
+
+ Keyword arguments:
+ delete_v1 -- delete any ID3v1 tag
+ delete_v2 -- delete any ID3v2 tag
+ """
+ if filename is None:
+ filename = self.filename
+ delete(filename, delete_v1, delete_v2)
+ self.clear()
+
+ def __save_frame(self, frame):
+ flags = 0
+ if self.PEDANTIC and isinstance(frame, TextFrame):
+ if len(str(frame)) == 0: return ''
+ framedata = frame._writeData()
+ usize = len(framedata)
+ if usize > 2048:
+ framedata = BitPaddedInt.to_str(usize) + framedata.encode('zlib')
+ flags |= Frame.FLAG24_COMPRESS | Frame.FLAG24_DATALEN
+ datasize = BitPaddedInt.to_str(len(framedata), width=4)
+ header = pack('>4s4sH', type(frame).__name__, datasize, flags)
+ return header + framedata
+
+ def update_to_v24(self):
+ """Convert older tags into an ID3v2.4 tag.
+
+ This updates old ID3v2 frames to ID3v2.4 ones (e.g. TYER to
+ TDRC). If you intend to save tags, you must call this function
+ at some point; it is called by default when loading the tag.
+ """
+
+ if self.version < (2,3,0): del self.unknown_frames[:]
+ # unsafe to write
+
+ # TDAT, TYER, and TIME have been turned into TDRC.
+ try:
+ if str(self.get("TYER", "")).strip("\x00"):
+ date = str(self.pop("TYER"))
+ if str(self.get("TDAT", "")).strip("\x00"):
+ dat = str(self.pop("TDAT"))
+ date = "%s-%s-%s" % (date, dat[2:], dat[:2])
+ if str(self.get("TIME", "")).strip("\x00"):
+ time = str(self.pop("TIME"))
+ date += "T%s:%s:00" % (time[:2], time[2:])
+ if "TDRC" not in self:
+ self.add(TDRC(encoding=0, text=date))
+ except UnicodeDecodeError:
+ # Old ID3 tags have *lots* of Unicode problems, so if TYER
+ # is bad, just chuck the frames.
+ pass
+
+ # TORY can be the first part of a TDOR.
+ if "TORY" in self:
+ f = self.pop("TORY")
+ if "TDOR" not in self:
+ try:
+ self.add(TDOR(encoding=0, text=str(f)))
+ except UnicodeDecodeError:
+ pass
+
+ # IPLS is now TIPL.
+ if "IPLS" in self:
+ f = self.pop("IPLS")
+ if "TIPL" not in self:
+ self.add(TIPL(encoding=f.encoding, people=f.people))
+
+ if "TCON" in self:
+ # Get rid of "(xx)Foobr" format.
+ self["TCON"].genres = self["TCON"].genres
+
+ if self.version < (2, 3):
+ # ID3v2.2 PIC frames are slightly different.
+ pics = self.getall("APIC")
+ mimes = { "PNG": "image/png", "JPG": "image/jpeg" }
+ self.delall("APIC")
+ for pic in pics:
+ newpic = APIC(
+ encoding=pic.encoding, mime=mimes.get(pic.mime, pic.mime),
+ type=pic.type, desc=pic.desc, data=pic.data)
+ self.add(newpic)
+
+ # ID3v2.2 LNK frames are just way too different to upgrade.
+ self.delall("LINK")
+
+ # These can't be trivially translated to any ID3v2.4 tags, or
+ # should have been removed already.
+ for key in ["RVAD", "EQUA", "TRDA", "TSIZ", "TDAT", "TIME", "CRM"]:
+ if key in self: del(self[key])
+
+def delete(filename, delete_v1=True, delete_v2=True):
+ """Remove tags from a file.
+
+ Keyword arguments:
+ delete_v1 -- delete any ID3v1 tag
+ delete_v2 -- delete any ID3v2 tag
+ """
+
+ f = open(filename, 'rb+')
+
+ if delete_v1:
+ try:
+ f.seek(-128, 2)
+ except IOError: pass
+ else:
+ if f.read(3) == "TAG":
+ f.seek(-128, 2)
+ f.truncate()
+
+ # technically an insize=0 tag is invalid, but we delete it anyway
+ # (primarily because we used to write it)
+ if delete_v2:
+ f.seek(0, 0)
+ idata = f.read(10)
+ try: id3, vmaj, vrev, flags, insize = unpack('>3sBBB4s', idata)
+ except struct.error: id3, insize = '', -1
+ insize = BitPaddedInt(insize)
+ if id3 == 'ID3' and insize >= 0:
+ delete_bytes(f, insize + 10, 0)
+
+class BitPaddedInt(int):
+ def __new__(cls, value, bits=7, bigendian=True):
+ "Strips 8-bits bits out of every byte"
+ mask = (1<<(bits))-1
+ if isinstance(value, (int, long)):
+ bytes = []
+ while value:
+ bytes.append(value & ((1<<bits)-1))
+ value = value >> 8
+ if isinstance(value, str):
+ bytes = [ord(byte) & mask for byte in value]
+ if bigendian: bytes.reverse()
+ numeric_value = 0
+ for shift, byte in zip(range(0, len(bytes)*bits, bits), bytes):
+ numeric_value += byte << shift
+ if isinstance(numeric_value, long):
+ self = long.__new__(BitPaddedLong, numeric_value)
+ else:
+ self = int.__new__(BitPaddedInt, numeric_value)
+ self.bits = bits
+ self.bigendian = bigendian
+ return self
+
+ def as_str(value, bits=7, bigendian=True, width=4):
+ bits = getattr(value, 'bits', bits)
+ bigendian = getattr(value, 'bigendian', bigendian)
+ value = int(value)
+ mask = (1<<bits)-1
+ bytes = []
+ while value:
+ bytes.append(value & mask)
+ value = value >> bits
+ # PCNT and POPM use growing integers of at least 4 bytes as counters.
+ if width == -1: width = max(4, len(bytes))
+ if len(bytes) > width:
+ raise ValueError, 'Value too wide (%d bytes)' % len(bytes)
+ else: bytes.extend([0] * (width-len(bytes)))
+ if bigendian: bytes.reverse()
+ return ''.join(map(chr, bytes))
+ to_str = staticmethod(as_str)
+
+class BitPaddedLong(long):
+ def as_str(value, bits=7, bigendian=True, width=4):
+ return BitPaddedInt.to_str(value, bits, bigendian, width)
+ to_str = staticmethod(as_str)
+
+class unsynch(object):
+ def decode(value):
+ output = []
+ safe = True
+ append = output.append
+ for val in value:
+ if safe:
+ append(val)
+ safe = val != '\xFF'
+ else:
+ if val >= '\xE0': raise ValueError('invalid sync-safe string')
+ elif val != '\x00': append(val)
+ safe = True
+ if not safe: raise ValueError('string ended unsafe')
+ return ''.join(output)
+ decode = staticmethod(decode)
+
+ def encode(value):
+ output = []
+ safe = True
+ append = output.append
+ for val in value:
+ if safe:
+ append(val)
+ if val == '\xFF': safe = False
+ elif val == '\x00' or val >= '\xE0':
+ append('\x00')
+ append(val)
+ safe = val != '\xFF'
+ else:
+ append(val)
+ safe = True
+ if not safe: append('\x00')
+ return ''.join(output)
+ encode = staticmethod(encode)
+
+class Spec(object):
+ def __init__(self, name): self.name = name
+ def __hash__(self): raise TypeError("Spec objects are unhashable")
+
+class ByteSpec(Spec):
+ def read(self, frame, data): return ord(data[0]), data[1:]
+ def write(self, frame, value): return chr(value)
+ def validate(self, frame, value): return value
+
+class IntegerSpec(Spec):
+ def read(self, frame, data):
+ return int(BitPaddedInt(data, bits=8)), ''
+ def write(self, frame, value):
+ return BitPaddedInt.to_str(value, bits=8, width=-1)
+ def validate(self, frame, value):
+ return value
+
+class SizedIntegerSpec(Spec):
+ def __init__(self, name, size):
+ self.name, self.__sz = name, size
+ def read(self, frame, data):
+ return int(BitPaddedInt(data[:self.__sz], bits=8)), data[self.__sz:]
+ def write(self, frame, value):
+ return BitPaddedInt.to_str(value, bits=8, width=self.__sz)
+ def validate(self, frame, value):
+ return value
+
+class EncodingSpec(ByteSpec):
+ def read(self, frame, data):
+ enc, data = super(EncodingSpec, self).read(frame, data)
+ if enc < 16: return enc, data
+ else: return 0, chr(enc)+data
+
+ def validate(self, frame, value):
+ if 0 <= value <= 3: return value
+ if value is None: return None
+ raise ValueError, 'Invalid Encoding: %r' % value
+
+class StringSpec(Spec):
+ def __init__(self, name, length):
+ super(StringSpec, self).__init__(name)
+ self.len = length
+ def read(s, frame, data): return data[:s.len], data[s.len:]
+ def write(s, frame, value):
+ if value is None: return '\x00' * s.len
+ else: return (str(value) + '\x00' * s.len)[:s.len]
+ def validate(s, frame, value):
+ if value is None: return None
+ if isinstance(value, basestring) and len(value) == s.len: return value
+ raise ValueError, 'Invalid StringSpec[%d] data: %r' % (s.len, value)
+
+class BinaryDataSpec(Spec):
+ def read(self, frame, data): return data, ''
+ def write(self, frame, value): return str(value)
+ def validate(self, frame, value): return str(value)
+
+class EncodedTextSpec(Spec):
+ # Okay, seriously. This is private and defined explicitly and
+ # completely by the ID3 specification. You can't just add
+ # encodings here however you want.
+ _encodings = ( ('latin1', '\x00'), ('utf16', '\x00\x00'),
+ ('utf_16_be', '\x00\x00'), ('utf8', '\x00') )
+
+ def read(self, frame, data):
+ enc, term = self._encodings[frame.encoding]
+ ret = ''
+ if len(term) == 1:
+ if term in data:
+ data, ret = data.split(term, 1)
+ else:
+ offset = -1
+ try:
+ while True:
+ offset = data.index(term, offset+1)
+ if offset & 1: continue
+ data, ret = data[0:offset], data[offset+2:]; break
+ except ValueError: pass
+
+ if len(data) < len(term): return u'', ret
+ return data.decode(enc), ret
+
+ def write(self, frame, value):
+ enc, term = self._encodings[frame.encoding]
+ return value.encode(enc) + term
+
+ def validate(self, frame, value): return unicode(value)
+
+class MultiSpec(Spec):
+ def __init__(self, name, *specs, **kw):
+ super(MultiSpec, self).__init__(name)
+ self.specs = specs
+ self.sep = kw.get('sep')
+
+ def read(self, frame, data):
+ values = []
+ while data:
+ record = []
+ for spec in self.specs:
+ value, data = spec.read(frame, data)
+ record.append(value)
+ if len(self.specs) != 1: values.append(record)
+ else: values.append(record[0])
+ return values, data
+
+ def write(self, frame, value):
+ data = []
+ if len(self.specs) == 1:
+ for v in value:
+ data.append(self.specs[0].write(frame, v))
+ else:
+ for record in value:
+ for v, s in zip(record, self.specs):
+ data.append(s.write(frame, v))
+ return ''.join(data)
+
+ def validate(self, frame, value):
+ if value is None: return []
+ if self.sep and isinstance(value, basestring):
+ value = value.split(self.sep)
+ if isinstance(value, list):
+ if len(self.specs) == 1:
+ return [self.specs[0].validate(frame, v) for v in value]
+ else:
+ return [
+ [s.validate(frame, v) for (v,s) in zip(val, self.specs)]
+ for val in value ]
+ raise ValueError, 'Invalid MultiSpec data: %r' % value
+
+class EncodedNumericTextSpec(EncodedTextSpec): pass
+class EncodedNumericPartTextSpec(EncodedTextSpec): pass
+
+class Latin1TextSpec(EncodedTextSpec):
+ def read(self, frame, data):
+ if '\x00' in data: data, ret = data.split('\x00',1)
+ else: ret = ''
+ return data.decode('latin1'), ret
+
+ def write(self, data, value):
+ return value.encode('latin1') + '\x00'
+
+ def validate(self, frame, value): return unicode(value)
+
+class ID3TimeStamp(object):
+ """A time stamp in ID3v2 format.
+
+ This is a restricted form of the ISO 8601 standard; time stamps
+ take the form of:
+ YYYY-MM-DD HH:MM:SS
+ Or some partial form (YYYY-MM-DD HH, YYYY, etc.).
+
+ The 'text' attribute contains the raw text data of the time stamp.
+ """
+
+ import re
+ def __init__(self, text):
+ if isinstance(text, ID3TimeStamp): text = text.text
+ self.text = text
+
+ __formats = ['%04d'] + ['%02d'] * 5
+ __seps = ['-', '-', ' ', ':', ':', 'x']
+ def get_text(self):
+ parts = [self.year, self.month, self.day,
+ self.hour, self.minute, self.second]
+ pieces = []
+ for i, part in enumerate(iter(iter(parts).next, None)):
+ pieces.append(self.__formats[i]%part + self.__seps[i])
+ return u''.join(pieces)[:-1]
+
+ def set_text(self, text, splitre=re.compile('[-T:/.]|\s+')):
+ year, month, day, hour, minute, second = \
+ splitre.split(text + ':::::')[:6]
+ for a in 'year month day hour minute second'.split():
+ try: v = int(locals()[a])
+ except ValueError: v = None
+ setattr(self, a, v)
+
+ text = property(get_text, set_text, doc="ID3v2.4 date and time.")
+
+ def __str__(self): return self.text
+ def __repr__(self): return repr(self.text)
+ def __cmp__(self, other): return cmp(self.text, other.text)
+ def encode(self, *args): return self.text.encode(*args)
+
+class TimeStampSpec(EncodedTextSpec):
+ def read(self, frame, data):
+ value, data = super(TimeStampSpec, self).read(frame, data)
+ return self.validate(frame, value), data
+
+ def write(self, frame, data):
+ return super(TimeStampSpec, self).write(frame,
+ data.text.replace(' ', 'T'))
+
+ def validate(self, frame, value):
+ try: return ID3TimeStamp(value)
+ except TypeError: raise ValueError, "Invalid ID3TimeStamp: %r" % value
+
+class ChannelSpec(ByteSpec):
+ (OTHER, MASTER, FRONTRIGHT, FRONTLEFT, BACKRIGHT, BACKLEFT, FRONTCENTRE,
+ BACKCENTRE, SUBWOOFER) = range(9)
+
+class VolumeAdjustmentSpec(Spec):
+ def read(self, frame, data):
+ value, = unpack('>h', data[0:2])
+ return value/512.0, data[2:]
+
+ def write(self, frame, value):
+ return pack('>h', int(round(value * 512)))
+
+ def validate(self, frame, value): return value
+
+class VolumePeakSpec(Spec):
+ def read(self, frame, data):
+ # http://bugs.xmms.org/attachment.cgi?id=113&action=view
+ peak = 0
+ bits = ord(data[0])
+ bytes = min(4, (bits + 7) >> 3)
+ # not enough frame data
+ if bytes + 1 > len(data): raise ID3JunkFrameError
+ shift = ((8 - (bits & 7)) & 7) + (4 - bytes) * 8
+ for i in range(1, bytes+1):
+ peak *= 256
+ peak += ord(data[i])
+ peak *= 2**shift
+ return (float(peak) / (2**31-1)), data[1+bytes:]
+
+ def write(self, frame, value):
+ # always write as 16 bits for sanity.
+ return "\x10" + pack('>H', int(round(value * 32768)))
+
+ def validate(self, frame, value): return value
+
+class SynchronizedTextSpec(EncodedTextSpec):
+ def read(self, frame, data):
+ texts = []
+ encoding, term = self._encodings[frame.encoding]
+ while data:
+ l = len(term)
+ value_idx = data.index(term)
+ value = data[:value_idx].decode(encoding)
+ time, = struct.unpack(">I", data[value_idx+l:value_idx+l+4])
+ texts.append((value, time))
+ data = data[value_idx+l+4:]
+ return texts, ""
+
+ def write(self, frame, value):
+ data = []
+ encoding, term = self._encodings[frame.encoding]
+ for text, time in frame.text:
+ text = text.encode(encoding) + term
+ data.append(text + struct.pack(">I", time))
+ return "".join(data)
+
+ def validate(self, frame, value):
+ return value
+
+class KeyEventSpec(Spec):
+ def read(self, frame, data):
+ events = []
+ while len(data) >= 5:
+ events.append(struct.unpack(">bI", data[:5]))
+ data = data[5:]
+ return events, data
+
+ def write(self, frame, value):
+ return "".join([struct.pack(">bI", *event) for event in value])
+
+ def validate(self, frame, value):
+ return value
+
+class VolumeAdjustmentsSpec(Spec):
+ # Not to be confused with VolumeAdjustmentSpec.
+ def read(self, frame, data):
+ adjustments = {}
+ while len(data) >= 4:
+ freq, adj = struct.unpack(">Hh", data[:4])
+ data = data[4:]
+ freq /= 2.0
+ adj /= 512.0
+ adjustments[freq] = adj
+ adjustments = adjustments.items()
+ adjustments.sort()
+ return adjustments, data
+
+ def write(self, frame, value):
+ value.sort()
+ return "".join([struct.pack(">Hh", int(freq * 2), int(adj * 512))
+ for (freq, adj) in value])
+
+ def validate(self, frame, value):
+ return value
+
+class ASPIIndexSpec(Spec):
+ def read(self, frame, data):
+ if frame.b == 16:
+ format = "H"
+ size = 2
+ elif frame.b == 8:
+ format = "B"
+ size = 1
+ else:
+ warn("invalid bit count in ASPI (%d)" % frame.b, ID3Warning)
+ return [], data
+
+ indexes = data[:frame.N * size]
+ data = data[frame.N * size:]
+ return list(struct.unpack(">" + format * frame.N, indexes)), data
+
+ def write(self, frame, values):
+ if frame.b == 16: format = "H"
+ elif frame.b == 8: format = "B"
+ else: raise ValueError("frame.b must be 8 or 16")
+ return struct.pack(">" + format * frame.N, *values)
+
+ def validate(self, frame, values):
+ return values
+
+class Frame(object):
+ """Fundamental unit of ID3 data.
+
+ ID3 tags are split into frames. Each frame has a potentially
+ different structure, and so this base class is not very featureful.
+ """
+
+ FLAG23_ALTERTAG = 0x8000
+ FLAG23_ALTERFILE = 0x4000
+ FLAG23_READONLY = 0x2000
+ FLAG23_COMPRESS = 0x0080
+ FLAG23_ENCRYPT = 0x0040
+ FLAG23_GROUP = 0x0020
+
+ FLAG24_ALTERTAG = 0x4000
+ FLAG24_ALTERFILE = 0x2000
+ FLAG24_READONLY = 0x1000
+ FLAG24_GROUPID = 0x0040
+ FLAG24_COMPRESS = 0x0008
+ FLAG24_ENCRYPT = 0x0004
+ FLAG24_UNSYNCH = 0x0002
+ FLAG24_DATALEN = 0x0001
+
+ _framespec = []
+ def __init__(self, *args, **kwargs):
+ if len(args)==1 and len(kwargs)==0 and isinstance(args[0], type(self)):
+ other = args[0]
+ for checker in self._framespec:
+ val = checker.validate(self, getattr(other, checker.name))
+ setattr(self, checker.name, val)
+ else:
+ for checker, val in zip(self._framespec, args):
+ setattr(self, checker.name, checker.validate(self, val))
+ for checker in self._framespec[len(args):]:
+ validated = checker.validate(
+ self, kwargs.get(checker.name, None))
+ setattr(self, checker.name, validated)
+
+ HashKey = property(
+ lambda s: s.FrameID,
+ doc="an internal key used to ensure frame uniqueness in a tag")
+ FrameID = property(
+ lambda s: type(s).__name__,
+ doc="ID3v2 three or four character frame ID")
+
+ def __repr__(self):
+ """Python representation of a frame.
+
+ The string returned is a valid Python expression to construct
+ a copy of this frame.
+ """
+ kw = []
+ for attr in self._framespec:
+ kw.append('%s=%r' % (attr.name, getattr(self, attr.name)))
+ return '%s(%s)' % (type(self).__name__, ', '.join(kw))
+
+ def _readData(self, data):
+ odata = data
+ for reader in self._framespec:
+ if len(data):
+ try: value, data = reader.read(self, data)
+ except UnicodeDecodeError:
+ raise ID3JunkFrameError
+ else: raise ID3JunkFrameError
+ setattr(self, reader.name, value)
+ if data.strip('\x00'):
+ warn('Leftover data: %s: %r (from %r)' % (
+ type(self).__name__, data, odata),
+ ID3Warning)
+
+ def _writeData(self):
+ data = []
+ for writer in self._framespec:
+ data.append(writer.write(self, getattr(self, writer.name)))
+ return ''.join(data)
+
+ def pprint(self):
+ """Return a human-readable representation of the frame."""
+ return "%s=%s" % (type(self).__name__, self._pprint())
+
+ def _pprint(self):
+ return "[unrepresentable data]"
+
+ def fromData(cls, id3, tflags, data):
+ """Construct this ID3 frame from raw string data."""
+
+ if (2,4,0) <= id3.version:
+ if tflags & (Frame.FLAG24_COMPRESS | Frame.FLAG24_DATALEN):
+ # The data length int is syncsafe in 2.4 (but not 2.3).
+ # However, we don't actually need the data length int,
+ # except to work around a QL 0.12 bug, and in that case
+ # all we need are the raw bytes.
+ datalen_bytes = data[:4]
+ data = data[4:]
+ if tflags & Frame.FLAG24_UNSYNCH or id3.f_unsynch:
+ try: data = unsynch.decode(data)
+ except ValueError, err:
+ if id3.PEDANTIC:
+ raise ID3BadUnsynchData, '%s: %r' % (err, data)
+ if tflags & Frame.FLAG24_ENCRYPT:
+ raise ID3EncryptionUnsupportedError
+ if tflags & Frame.FLAG24_COMPRESS:
+ try: data = data.decode('zlib')
+ except zlibError, err:
+ # the initial mutagen that went out with QL 0.12 did not
+ # write the 4 bytes of uncompressed size. Compensate.
+ data = datalen_bytes + data
+ try: data = data.decode('zlib')
+ except zlibError, err:
+ if id3.PEDANTIC:
+ raise ID3BadCompressedData, '%s: %r' % (err, data)
+
+ elif (2,3,0) <= id3.version:
+ if tflags & Frame.FLAG23_COMPRESS:
+ usize, = unpack('>L', data[:4])
+ data = data[4:]
+ if tflags & Frame.FLAG23_ENCRYPT:
+ raise ID3EncryptionUnsupportedError
+ if tflags & Frame.FLAG23_COMPRESS:
+ try: data = data.decode('zlib')
+ except zlibError, err:
+ if id3.PEDANTIC:
+ raise ID3BadCompressedData, '%s: %r' % (err, data)
+
+ frame = cls()
+ frame._rawdata = data
+ frame._flags = tflags
+ frame._readData(data)
+ return frame
+ fromData = classmethod(fromData)
+
+ def __hash__(self):
+ raise TypeError("Frame objects are unhashable")
+
+class FrameOpt(Frame):
+ """A frame with optional parts.
+
+ Some ID3 frames have optional data; this class extends Frame to
+ provide support for those parts.
+ """
+ _optionalspec = []
+
+ def __init__(self, *args, **kwargs):
+ super(FrameOpt, self).__init__(*args, **kwargs)
+ for spec in self._optionalspec:
+ if spec.name in kwargs:
+ validated = spec.validate(self, kwargs[spec.name])
+ setattr(self, spec.name, validated)
+ else: break
+
+ def _readData(self, data):
+ odata = data
+ for reader in self._framespec:
+ if len(data): value, data = reader.read(self, data)
+ else: raise ID3JunkFrameError
+ setattr(self, reader.name, value)
+ if data:
+ for reader in self._optionalspec:
+ if len(data): value, data = reader.read(self, data)
+ else: break
+ setattr(self, reader.name, value)
+ if data.strip('\x00'):
+ warn('Leftover data: %s: %r (from %r)' % (
+ type(self).__name__, data, odata),
+ ID3Warning)
+
+ def _writeData(self):
+ data = []
+ for writer in self._framespec:
+ data.append(writer.write(self, getattr(self, writer.name)))
+ for writer in self._optionalspec:
+ try: data.append(writer.write(self, getattr(self, writer.name)))
+ except AttributeError: break
+ return ''.join(data)
+
+ def __repr__(self):
+ kw = []
+ for attr in self._framespec:
+ kw.append('%s=%r' % (attr.name, getattr(self, attr.name)))
+ for attr in self._optionalspec:
+ if hasattr(self, attr.name):
+ kw.append('%s=%r' % (attr.name, getattr(self, attr.name)))
+ return '%s(%s)' % (type(self).__name__, ', '.join(kw))
+
+
+class TextFrame(Frame):
+ """Text strings.
+
+ Text frames support casts to unicode or str objects, as well as
+ list-like indexing, extend, and append.
+
+ Iterating over a TextFrame iterates over its strings, not its
+ characters.
+
+ Text frames have a 'text' attribute which is the list of strings,
+ and an 'encoding' attribute; 0 for ISO-8859 1, 1 UTF-16, 2 for
+ UTF-16BE, and 3 for UTF-8. If you don't want to worry about
+ encodings, just set it to 3.
+ """
+
+ _framespec = [ EncodingSpec('encoding'),
+ MultiSpec('text', EncodedTextSpec('text'), sep=u'\u0000') ]
+ def __str__(self): return self.__unicode__().encode('utf-8')
+ def __unicode__(self): return u'\u0000'.join(self.text)
+ def __eq__(self, other):
+ if isinstance(other, str): return str(self) == other
+ elif isinstance(other, unicode):
+ return u'\u0000'.join(self.text) == other
+ return self.text == other
+ def __getitem__(self, item): return self.text[item]
+ def __iter__(self): return iter(self.text)
+ def append(self, value): return self.text.append(value)
+ def extend(self, value): return self.text.extend(value)
+ def _pprint(self): return " / ".join(self.text)
+
+class NumericTextFrame(TextFrame):
+ """Numerical text strings.
+
+ The numeric value of these frames can be gotten with unary plus, e.g.
+ frame = TLEN('12345')
+ length = +frame
+ """
+
+ _framespec = [ EncodingSpec('encoding'),
+ MultiSpec('text', EncodedNumericTextSpec('text'), sep=u'\u0000') ]
+
+ def __pos__(self):
+ """Return the numerical value of the string."""
+ return int(self.text[0])
+
+class NumericPartTextFrame(TextFrame):
+ """Multivalue numerical text strings.
+
+ These strings indicate 'part (e.g. track) X of Y', and unary plus
+ returns the first value:
+ frame = TRCK('4/15')
+ track = +frame # track == 4
+ """
+
+ _framespec = [ EncodingSpec('encoding'),
+ MultiSpec('text', EncodedNumericPartTextSpec('text'), sep=u'\u0000') ]
+ def __pos__(self):
+ return int(self.text[0].split("/")[0])
+
+class TimeStampTextFrame(TextFrame):
+ """A list of time stamps.
+
+ The 'text' attribute in this frame is a list of ID3TimeStamp
+ objects, not a list of strings.
+ """
+
+ _framespec = [ EncodingSpec('encoding'),
+ MultiSpec('text', TimeStampSpec('stamp'), sep=u',') ]
+ def __str__(self): return self.__unicode__().encode('utf-8')
+ def __unicode__(self): return ','.join([stamp.text for stamp in self.text])
+ def _pprint(self):
+ return " / ".join([stamp.text for stamp in self.text])
+
+class UrlFrame(Frame):
+ """A frame containing a URL string.
+
+ The ID3 specification is silent about IRIs and normalized URL
+ forms. Mutagen assumes all URLs in files are encoded as Latin 1,
+ but string conversion of this frame returns a UTF-8 representation
+ for compatibility with other string conversions.
+
+ The only sane way to handle URLs in MP3s is to restrict them to
+ ASCII.
+ """
+
+ _framespec = [ Latin1TextSpec('url') ]
+ def __str__(self): return self.url.encode('utf-8')
+ def __unicode__(self): return self.url
+ def __eq__(self, other): return self.url == other
+ def _pprint(self): return self.url
+
+class UrlFrameU(UrlFrame):
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.url))
+
+class TALB(TextFrame): "Album"
+class TBPM(NumericTextFrame): "Beats per minute"
+class TCOM(TextFrame): "Composer"
+
+class TCON(TextFrame):
+ """Content type (Genre)
+
+ ID3 has several ways genres can be represented; for convenience,
+ use the 'genres' property rather than the 'text' attribute.
+ """
+
+ from mutagen._constants import GENRES
+
+ def __get_genres(self):
+ genres = []
+ import re
+ genre_re = re.compile(r"((?:\((?P<id>[0-9]+|RX|CR)\))*)(?P<str>.+)?")
+ for value in self.text:
+ if value.isdigit():
+ try: genres.append(self.GENRES[int(value)])
+ except IndexError: genres.append(u"Unknown")
+ elif value == "CR": genres.append(u"Cover")
+ elif value == "RX": genres.append(u"Remix")
+ elif value:
+ newgenres = []
+ genreid, dummy, genrename = genre_re.match(value).groups()
+
+ if genreid:
+ for gid in genreid[1:-1].split(")("):
+ if gid.isdigit() and int(gid) < len(self.GENRES):
+ gid = unicode(self.GENRES[int(gid)])
+ newgenres.append(gid)
+ elif gid == "CR": newgenres.append(u"Cover")
+ elif gid == "RX": newgenres.append(u"Remix")
+ else: newgenres.append(u"Unknown")
+
+ if genrename:
+ # "Unescaping" the first parenthesis
+ if genrename.startswith("(("): genrename = genrename[1:]
+ if genrename not in newgenres: newgenres.append(genrename)
+
+ genres.extend(newgenres)
+
+ return genres
+
+ def __set_genres(self, genres):
+ if isinstance(genres, basestring): genres = [genres]
+ self.text = map(self.__decode, genres)
+
+ def __decode(self, value):
+ if isinstance(value, str):
+ enc = EncodedTextSpec._encodings[self.encoding][0]
+ return value.decode(enc)
+ else: return value
+
+ genres = property(__get_genres, __set_genres, None,
+ "A list of genres parsed from the raw text data.")
+
+ def _pprint(self):
+ return " / ".join(self.genres)
+
+class TCOP(TextFrame): "Copyright (c)"
+class TCMP(NumericTextFrame): "iTunes Compilation Flag"
+class TDAT(TextFrame): "Date of recording (DDMM)"
+class TDEN(TimeStampTextFrame): "Encoding Time"
+class TDOR(TimeStampTextFrame): "Original Release Time"
+class TDLY(NumericTextFrame): "Audio Delay (ms)"
+class TDRC(TimeStampTextFrame): "Recording Time"
+class TDRL(TimeStampTextFrame): "Release Time"
+class TDTG(TimeStampTextFrame): "Tagging Time"
+class TENC(TextFrame): "Encoder"
+class TEXT(TextFrame): "Lyricist"
+class TFLT(TextFrame): "File type"
+class TIME(TextFrame): "Time of recording (HHMM)"
+class TIT1(TextFrame): "Content group description"
+class TIT2(TextFrame): "Title"
+class TIT3(TextFrame): "Subtitle/Description refinement"
+class TKEY(TextFrame): "Starting Key"
+class TLAN(TextFrame): "Audio Languages"
+class TLEN(NumericTextFrame): "Audio Length (ms)"
+class TMED(TextFrame): "Source Media Type"
+class TMOO(TextFrame): "Mood"
+class TOAL(TextFrame): "Original Album"
+class TOFN(TextFrame): "Original Filename"
+class TOLY(TextFrame): "Original Lyricist"
+class TOPE(TextFrame): "Original Artist/Performer"
+class TORY(NumericTextFrame): "Original Release Year"
+class TOWN(TextFrame): "Owner/Licensee"
+class TPE1(TextFrame): "Lead Artist/Performer/Soloist/Group"
+class TPE2(TextFrame): "Band/Orchestra/Accompaniment"
+class TPE3(TextFrame): "Conductor"
+class TPE4(TextFrame): "Interpreter/Remixer/Modifier"
+class TPOS(NumericPartTextFrame): "Part of set"
+class TPRO(TextFrame): "Produced (P)"
+class TPUB(TextFrame): "Publisher"
+class TRCK(NumericPartTextFrame): "Track Number"
+class TRDA(TextFrame): "Recording Dates"
+class TRSN(TextFrame): "Internet Radio Station Name"
+class TRSO(TextFrame): "Internet Radio Station Owner"
+class TSIZ(NumericTextFrame): "Size of audio data (bytes)"
+class TSOA(TextFrame): "Album Sort Order key"
+class TSOP(TextFrame): "Perfomer Sort Order key"
+class TSOT(TextFrame): "Title Sort Order key"
+class TSRC(TextFrame): "International Standard Recording Code (ISRC)"
+class TSSE(TextFrame): "Encoder settings"
+class TSST(TextFrame): "Set Subtitle"
+class TYER(NumericTextFrame): "Year of recording"
+
+class TXXX(TextFrame):
+ """User-defined text data.
+
+ TXXX frames have a 'desc' attribute which is set to any Unicode
+ value (though the encoding of the text and the description must be
+ the same). Many taggers use this frame to store freeform keys.
+ """
+ _framespec = [ EncodingSpec('encoding'), EncodedTextSpec('desc'),
+ MultiSpec('text', EncodedTextSpec('text'), sep=u'\u0000') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+ def _pprint(self): return "%s=%s" % (self.desc, " / ".join(self.text))
+
+class WCOM(UrlFrameU): "Commercial Information"
+class WCOP(UrlFrame): "Copyright Information"
+class WOAF(UrlFrame): "Official File Information"
+class WOAR(UrlFrameU): "Official Artist/Performer Information"
+class WOAS(UrlFrame): "Official Source Information"
+class WORS(UrlFrame): "Official Internet Radio Information"
+class WPAY(UrlFrame): "Payment Information"
+class WPUB(UrlFrame): "Official Publisher Information"
+
+class WXXX(UrlFrame):
+ """User-defined URL data.
+
+ Like TXXX, this has a freeform description associated with it.
+ """
+ _framespec = [ EncodingSpec('encoding'), EncodedTextSpec('desc'),
+ Latin1TextSpec('url') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+
+class PairedTextFrame(Frame):
+ """Paired text strings.
+
+ Some ID3 frames pair text strings, to associate names with a more
+ specific involvement in the song. The 'people' attribute of these
+ frames contains a list of pairs:
+ [['trumpet', 'Miles Davis'], ['bass', 'Paul Chambers']]
+
+ Like text frames, these frames also have an encoding attribute.
+ """
+
+ _framespec = [ EncodingSpec('encoding'), MultiSpec('people',
+ EncodedTextSpec('involvement'), EncodedTextSpec('person')) ]
+ def __eq__(self, other):
+ return self.people == other
+
+class TIPL(PairedTextFrame): "Involved People List"
+class TMCL(PairedTextFrame): "Musicians Credits List"
+class IPLS(TIPL): "Involved People List"
+
+class MCDI(Frame):
+ """Binary dump of CD's TOC.
+
+ The 'data' attribute contains the raw byte string.
+ """
+ _framespec = [ BinaryDataSpec('data') ]
+ def __eq__(self, other): return self.data == other
+
+class ETCO(Frame):
+ """Event timing codes."""
+ _framespec = [ ByteSpec("format"), KeyEventSpec("events") ]
+ def __eq__(self, other): return self.events == other
+
+class MLLT(Frame):
+ """MPEG location lookup table.
+
+ This frame's attributes may be changed in the future based on
+ feedback from real-world use.
+ """
+ _framespec = [ SizedIntegerSpec('frames', 2),
+ SizedIntegerSpec('bytes', 3),
+ SizedIntegerSpec('milliseconds', 3),
+ ByteSpec('bits_for_bytes'),
+ ByteSpec('bits_for_milliseconds'),
+ BinaryDataSpec('data') ]
+ def __eq__(self, other): return self.data == other
+
+class SYTC(Frame):
+ """Synchronised tempo codes.
+
+ This frame's attributes may be changed in the future based on
+ feedback from real-world use.
+ """
+ _framespec = [ ByteSpec("format"), BinaryDataSpec("data") ]
+ def __eq__(self, other): return self.data == other
+
+class USLT(Frame):
+ """Unsynchronised lyrics/text transcription.
+
+ Lyrics have a three letter ISO language code ('lang'), a
+ description ('desc'), and a block of plain text ('text').
+ """
+
+ _framespec = [ EncodingSpec('encoding'), StringSpec('lang', 3),
+ EncodedTextSpec('desc'), EncodedTextSpec('text') ]
+ HashKey = property(lambda s: '%s:%s:%r' % (s.FrameID, s.desc, s.lang))
+
+ def __str__(self): return self.text.encode('utf-8')
+ def __unicode__(self): return self.text
+ def __eq__(self, other): return self.text == other
+
+class SYLT(Frame):
+ """Synchronised lyrics/text."""
+
+ _framespec = [ EncodingSpec('encoding'), StringSpec('lang', 3),
+ ByteSpec('format'), ByteSpec('type'), EncodedTextSpec('desc'),
+ SynchronizedTextSpec('text') ]
+ HashKey = property(lambda s: '%s:%s:%r' % (s.FrameID, s.desc, s.lang))
+
+ def __eq__(self, other):
+ return str(self) == other
+
+ def __str__(self):
+ return "".join([text for (text, time) in self.text]).encode('utf-8')
+
+class COMM(TextFrame):
+ """User comment.
+
+ User comment frames have a descrption, like TXXX, and also a three
+ letter ISO language code in the 'lang' attribute.
+ """
+ _framespec = [ EncodingSpec('encoding'), StringSpec('lang', 3),
+ EncodedTextSpec('desc'),
+ MultiSpec('text', EncodedTextSpec('text'), sep=u'\u0000') ]
+ HashKey = property(lambda s: '%s:%s:%r' % (s.FrameID, s.desc, s.lang))
+ def _pprint(self): return "%s=%r=%s" % (
+ self.desc, self.lang, " / ".join(self.text))
+
+class RVA2(Frame):
+ """Relative volume adjustment (2).
+
+ This frame is used to implemented volume scaling, and in
+ particular, normalization using ReplayGain.
+
+ Attributes:
+ desc -- description or context of this adjustment
+ channel -- audio channel to adjust (master is 1)
+ gain -- a + or - dB gain relative to some reference level
+ peak -- peak of the audio as a floating point number, [0, 1]
+
+ When storing ReplayGain tags, use descriptions of 'album' and
+ 'track' on channel 1.
+ """
+
+ _framespec = [ Latin1TextSpec('desc'), ChannelSpec('channel'),
+ VolumeAdjustmentSpec('gain'), VolumePeakSpec('peak') ]
+ _channels = ["Other", "Master volume", "Front right", "Front left",
+ "Back right", "Back left", "Front centre", "Back centre",
+ "Subwoofer"]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+
+ def __eq__(self, other):
+ return ((str(self) == other) or
+ (self.desc == other.desc and
+ self.channel == other.channel and
+ self.gain == other.gain and
+ self.peak == other.peak))
+
+ def __str__(self):
+ return "%s: %+0.4f dB/%0.4f" % (
+ self._channels[self.channel], self.gain, self.peak)
+
+class EQU2(Frame):
+ """Equalisation (2).
+
+ Attributes:
+ method -- interpolation method (0 = band, 1 = linear)
+ desc -- identifying description
+ adjustments -- list of (frequency, vol_adjustment) pairs
+ """
+ _framespec = [ ByteSpec("method"), Latin1TextSpec("desc"),
+ VolumeAdjustmentsSpec("adjustments") ]
+ def __eq__(self, other): return self.adjustments == other
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+
+# class RVAD: unsupported
+# class EQUA: unsupported
+
+class RVRB(Frame):
+ """Reverb."""
+ _framespec = [ SizedIntegerSpec('left', 2), SizedIntegerSpec('right', 2),
+ ByteSpec('bounce_left'), ByteSpec('bounce_right'),
+ ByteSpec('feedback_ltl'), ByteSpec('feedback_ltr'),
+ ByteSpec('feedback_rtr'), ByteSpec('feedback_rtl'),
+ ByteSpec('premix_ltr'), ByteSpec('premix_rtl') ]
+
+ def __eq__(self, other): return (self.left, self.right) == other
+
+class APIC(Frame):
+ """Attached (or linked) Picture.
+
+ Attributes:
+ encoding -- text encoding for the description
+ mime -- a MIME type (e.g. image/jpeg) or '-->' if the data is a URI
+ type -- the source of the image (3 is the album front cover)
+ desc -- a text description of the image
+ data -- raw image data, as a byte string
+
+ Mutagen will automatically compress large images when saving tags.
+ """
+ _framespec = [ EncodingSpec('encoding'), Latin1TextSpec('mime'),
+ ByteSpec('type'), EncodedTextSpec('desc'), BinaryDataSpec('data') ]
+ def __eq__(self, other): return self.data == other
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+ def _pprint(self):
+ return "%s (%s, %d bytes)" % (
+ self.desc, self.mime, len(self.data))
+
+class PCNT(Frame):
+ """Play counter.
+
+ The 'count' attribute contains the (recorded) number of times this
+ file has been played.
+
+ This frame is basically obsoleted by POPM.
+ """
+ _framespec = [ IntegerSpec('count') ]
+
+ def __eq__(self, other): return self.count == other
+ def __pos__(self): return self.count
+ def _pprint(self): return unicode(self.count)
+
+class POPM(Frame):
+ """Popularimeter.
+
+ This frame keys a rating (out of 255) and a play count to an email
+ address.
+
+ Attributes:
+ email -- email this POPM frame is for
+ rating -- rating from 0 to 255
+ count -- number of times the files has been played
+ """
+ _framespec = [ Latin1TextSpec('email'), ByteSpec('rating'),
+ IntegerSpec('count') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.email))
+
+ def __eq__(self, other): return self.rating == other
+ def __pos__(self): return self.rating
+ def _pprint(self): return "%s=%s %s/255" % (
+ self.email, self.count, self.rating)
+
+class GEOB(Frame):
+ """General Encapsulated Object.
+
+ A blob of binary data, that is not a picture (those go in APIC).
+
+ Attributes:
+ encoding -- encoding of the description
+ mime -- MIME type of the data or '-->' if the data is a URI
+ filename -- suggested filename if extracted
+ desc -- text description of the data
+ data -- raw data, as a byte string
+ """
+ _framespec = [ EncodingSpec('encoding'), Latin1TextSpec('mime'),
+ EncodedTextSpec('filename'), EncodedTextSpec('desc'),
+ BinaryDataSpec('data') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.desc))
+
+ def __eq__(self, other): return self.data == other
+
+class RBUF(FrameOpt):
+ """Recommended buffer size.
+
+ Attributes:
+ size -- recommended buffer size in bytes
+ info -- if ID3 tags may be elsewhere in the file (optional)
+ offset -- the location of the next ID3 tag, if any
+
+ Mutagen will not find the next tag itself.
+ """
+ _framespec = [ SizedIntegerSpec('size', 3) ]
+ _optionalspec = [ ByteSpec('info'), SizedIntegerSpec('offset', 4) ]
+
+ def __eq__(self, other): return self.size == other
+ def __pos__(self): return self.size
+
+class AENC(FrameOpt):
+ """Audio encryption.
+
+ Attributes:
+ owner -- key identifying this encryption type
+ preview_start -- unencrypted data block offset
+ preview_length -- number of unencrypted blocks
+ data -- data required for decryption (optional)
+
+ Mutagen cannot decrypt files.
+ """
+ _framespec = [ Latin1TextSpec('owner'),
+ SizedIntegerSpec('preview_start', 2),
+ SizedIntegerSpec('preview_length', 2) ]
+ _optionalspec = [ BinaryDataSpec('data') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.owner))
+
+ def __str__(self): return self.owner.encode('utf-8')
+ def __unicode__(self): return self.owner
+ def __eq__(self, other): return self.owner == other
+
+class LINK(FrameOpt):
+ """Linked information.
+
+ Attributes:
+ frameid -- the ID of the linked frame
+ url -- the location of the linked frame
+ data -- further ID information for the frame
+ """
+
+ _framespec = [ StringSpec('frameid', 4), Latin1TextSpec('url') ]
+ _optionalspec = [ BinaryDataSpec('data') ]
+ def __HashKey(self):
+ try:
+ return "%s:%s:%s:%r" % (
+ self.FrameID, self.frameid, self.url, self.data)
+ except AttributeError:
+ return "%s:%s:%s" % (self.FrameID, self.frameid, self.url)
+ HashKey = property(__HashKey)
+ def __eq__(self, other):
+ try: return (self.frameid, self.url, self.data) == other
+ except AttributeError: return (self.frameid, self.url) == other
+
+class POSS(Frame):
+ """Position synchronisation frame
+
+ Attribute:
+ format -- format of the position attribute (frames or milliseconds)
+ position -- current position of the file
+ """
+ _framespec = [ ByteSpec('format'), IntegerSpec('position') ]
+
+ def __pos__(self): return self.position
+ def __eq__(self, other): return self.position == other
+
+class UFID(Frame):
+ """Unique file identifier.
+
+ Attributes:
+ owner -- format/type of identifier
+ data -- identifier
+ """
+
+ _framespec = [ Latin1TextSpec('owner'), BinaryDataSpec('data') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.owner))
+ def __eq__(s, o):
+ if isinstance(o, UFI): return s.owner == o.owner and s.data == o.data
+ else: return s.data == o
+ def _pprint(self):
+ isascii = ord(max(self.data)) < 128
+ if isascii: return "%s=%s" % (self.owner, self.data)
+ else: return "%s (%d bytes)" % (self.owner, len(self.data))
+
+class USER(Frame):
+ """Terms of use.
+
+ Attributes:
+ encoding -- text encoding
+ lang -- ISO three letter language code
+ text -- licensing terms for the audio
+ """
+ _framespec = [ EncodingSpec('encoding'), StringSpec('lang', 3),
+ EncodedTextSpec('text') ]
+ HashKey = property(lambda s: '%s:%r' % (s.FrameID, s.lang))
+
+ def __str__(self): return self.text.encode('utf-8')
+ def __unicode__(self): return self.text
+ def __eq__(self, other): return self.text == other
+ def _pprint(self): return "%r=%s" % (self.lang, self.text)
+
+class OWNE(Frame):
+ """Ownership frame."""
+ _framespec = [ EncodingSpec('encoding'), Latin1TextSpec('price'),
+ StringSpec('date', 8), EncodedTextSpec('seller') ]
+
+ def __str__(self): return self.seller.encode('utf-8')
+ def __unicode__(self): return self.seller
+ def __eq__(self, other): return self.seller == other
+
+class COMR(FrameOpt):
+ """Commercial frame."""
+ _framespec = [ EncodingSpec('encoding'), Latin1TextSpec('price'),
+ StringSpec('valid_until', 8), Latin1TextSpec('contact'),
+ ByteSpec('format'), EncodedTextSpec('seller'),
+ EncodedTextSpec('desc')]
+ _optionalspec = [ Latin1TextSpec('mime'), BinaryDataSpec('logo') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s._writeData()))
+ def __eq__(self, other): return self._writeData() == other._writeData()
+
+class ENCR(Frame):
+ """Encryption method registration.
+
+ The standard does not allow multiple ENCR frames with the same owner
+ or the same method. Mutagen only verifies that the owner is unique.
+ """
+ _framespec = [ Latin1TextSpec('owner'), ByteSpec('method'),
+ BinaryDataSpec('data') ]
+ HashKey = property(lambda s: "%s:%s" % (s.FrameID, s.owner))
+ def __str__(self): return self.data
+ def __eq__(self, other): return self.data == other
+
+class GRID(FrameOpt):
+ """Group identification registration."""
+ _framespec = [ Latin1TextSpec('owner'), ByteSpec('group') ]
+ _optionalspec = [ BinaryDataSpec('data') ]
+ HashKey = property(lambda s: '%s:%s' % (s.FrameID, s.group))
+ def __pos__(self): return self.group
+ def __str__(self): return self.owner.encode('utf-8')
+ def __unicode__(self): return self.owner
+ def __eq__(self, other): return self.owner == other or self.group == other
+
+
+class PRIV(Frame):
+ """Private frame."""
+ _framespec = [ Latin1TextSpec('owner'), BinaryDataSpec('data') ]
+ HashKey = property(lambda s: '%s:%s:%s' % (
+ s.FrameID, s.owner, s.data.decode('latin1')))
+ def __str__(self): return self.data
+ def __eq__(self, other): return self.data == other
+ def _pprint(self):
+ isascii = ord(max(self.data)) < 128
+ if isascii: return "%s=%s" % (self.owner, self.data)
+ else: return "%s (%d bytes)" % (self.owner, len(self.data))
+
+class SIGN(Frame):
+ """Signature frame."""
+ _framespec = [ ByteSpec('group'), BinaryDataSpec('sig') ]
+ HashKey = property(lambda s: '%s:%c:%s' % (s.FrameID, s.group, s.sig))
+ def __str__(self): return self.sig
+ def __eq__(self, other): return self.sig == other
+
+class SEEK(Frame):
+ """Seek frame.
+
+ Mutagen does not find tags at seek offsets.
+ """
+ _framespec = [ IntegerSpec('offset') ]
+ def __pos__(self): return self.offset
+ def __eq__(self, other): return self.offset == other
+
+class ASPI(Frame):
+ """Audio seek point index.
+
+ Attributes: S, L, N, b, and Fi. For the meaning of these, see
+ the ID3v2.4 specification. Fi is a list of integers.
+ """
+ _framespec = [ SizedIntegerSpec("S", 4), SizedIntegerSpec("L", 4),
+ SizedIntegerSpec("N", 2), ByteSpec("b"),
+ ASPIIndexSpec("Fi") ]
+ def __eq__(self, other): return self.Fi == other
+
+Frames = dict([(k,v) for (k,v) in globals().items()
+ if len(k)==4 and isinstance(v, type) and issubclass(v, Frame)])
+"""All supported ID3v2 frames, keyed by frame name."""
+del(k); del(v)
+
+# ID3v2.2 frames
+class UFI(UFID): "Unique File Identifier"
+
+class TT1(TIT1): "Content group description"
+class TT2(TIT2): "Title"
+class TT3(TIT3): "Subtitle/Description refinement"
+class TP1(TPE1): "Lead Artist/Performer/Soloist/Group"
+class TP2(TPE2): "Band/Orchestra/Accompaniment"
+class TP3(TPE3): "Conductor"
+class TP4(TPE4): "Interpreter/Remixer/Modifier"
+class TCM(TCOM): "Composer"
+class TXT(TEXT): "Lyricist"
+class TLA(TLAN): "Audio Language(s)"
+class TCO(TCON): "Content Type (Genre)"
+class TAL(TALB): "Album"
+class TPA(TPOS): "Part of set"
+class TRK(TRCK): "Track Number"
+class TRC(TSRC): "International Standard Recording Code (ISRC)"
+class TYE(TYER): "Year of recording"
+class TDA(TDAT): "Date of recording (DDMM)"
+class TIM(TIME): "Time of recording (HHMM)"
+class TRD(TRDA): "Recording Dates"
+class TMT(TMED): "Source Media Type"
+class TFT(TFLT): "File Type"
+class TBP(TBPM): "Beats per minute"
+class TCP(TCMP): "iTunes Compilation Flag"
+class TCR(TCOP): "Copyright (C)"
+class TPB(TPUB): "Publisher"
+class TEN(TENC): "Encoder"
+class TSS(TSSE): "Encoder settings"
+class TOF(TOFN): "Original Filename"
+class TLE(TLEN): "Audio Length (ms)"
+class TSI(TSIZ): "Audio Data size (bytes)"
+class TDY(TDLY): "Audio Delay (ms)"
+class TKE(TKEY): "Starting Key"
+class TOT(TOAL): "Original Album"
+class TOA(TOPE): "Original Artist/Perfomer"
+class TOL(TOLY): "Original Lyricist"
+class TOR(TORY): "Original Release Year"
+
+class TXX(TXXX): "User-defined Text"
+
+class WAF(WOAF): "Official File Information"
+class WAR(WOAR): "Official Artist/Performer Information"
+class WAS(WOAS): "Official Source Information"
+class WCM(WCOM): "Commercial Information"
+class WCP(WCOP): "Copyright Information"
+class WPB(WPUB): "Official Publisher Information"
+
+class WXX(WXXX): "User-defined URL"
+
+class IPL(IPLS): "Involved people list"
+class MCI(MCDI): "Binary dump of CD's TOC"
+class ETC(ETCO): "Event timing codes"
+class MLL(MLLT): "MPEG location lookup table"
+class STC(SYTC): "Synced tempo codes"
+class ULT(USLT): "Unsychronised lyrics/text transcription"
+class SLT(SYLT): "Synchronised lyrics/text"
+class COM(COMM): "Comment"
+#class RVA(RVAD)
+#class EQU(EQUA)
+class REV(RVRB): "Reverb"
+class PIC(APIC):
+ """Attached Picture.
+
+ The 'mime' attribute of an ID3v2.2 attached picture must be either
+ 'PNG' or 'JPG'.
+ """
+ _framespec = [ EncodingSpec('encoding'), StringSpec('mime', 3),
+ ByteSpec('type'), EncodedTextSpec('desc'), BinaryDataSpec('data') ]
+class GEO(GEOB): "General Encapsulated Object"
+class CNT(PCNT): "Play counter"
+class POP(POPM): "Popularimeter"
+class BUF(RBUF): "Recommended buffer size"
+
+class CRM(Frame):
+ """Encrypted meta frame"""
+ _framespec = [ Latin1TextSpec('owner'), Latin1TextSpec('desc'),
+ BinaryDataSpec('data') ]
+ def __eq__(self, other): return self.data == other
+
+class CRA(AENC): "Audio encryption"
+
+class LNK(LINK):
+ """Linked information"""
+ _framespec = [ StringSpec('frameid', 3), Latin1TextSpec('url') ]
+ _optionalspec = [ BinaryDataSpec('data') ]
+
+Frames_2_2 = dict([(k,v) for (k,v) in globals().items()
+ if len(k)==3 and isinstance(v, type) and issubclass(v, Frame)])
+
+# support open(filename) as interface
+Open = ID3
+
+# ID3v1.1 support.
+def ParseID3v1(string):
+ """Parse an ID3v1 tag, returning a list of ID3v2.4 frames."""
+ from struct import error as StructError
+ frames = {}
+ try:
+ tag, title, artist, album, year, comment, track, genre = unpack(
+ "3s30s30s30s4s29sBB", string)
+ except StructError: return None
+
+ if tag != "TAG": return None
+ def fix(string):
+ return string.split("\x00")[0].strip().decode('latin1')
+ title, artist, album, year, comment = map(
+ fix, [title, artist, album, year, comment])
+
+ if title: frames["TIT2"] = TIT2(encoding=0, text=title)
+ if artist: frames["TPE1"] = TPE1(encoding=0, text=[artist])
+ if album: frames["TALB"] = TALB(encoding=0, text=album)
+ if year: frames["TDRC"] = TDRC(encoding=0, text=year)
+ if comment: frames["COMM"] = COMM(
+ encoding=0, lang="eng", desc="ID3v1 Comment", text=comment)
+ # Don't read a track number if it looks like the comment was
+ # padded with spaces instead of nulls (thanks, WinAmp).
+ if track and (track != 32 or string[-3] == '\x00'):
+ frames["TRCK"] = TRCK(encoding=0, text=str(track))
+ if genre != 255: frames["TCON"] = TCON(encoding=0, text=str(genre))
+ return frames
+
+def MakeID3v1(id3):
+ """Return an ID3v1.1 tag string from a dict of ID3v2.4 frames."""
+
+ v1 = {}
+
+ for v2id, name in {"TIT2": "title", "TPE1": "artist",
+ "TALB": "album"}.items():
+ if v2id in id3:
+ text = id3[v2id].text[0].encode('latin1', 'replace')[:30]
+ else: text = ""
+ v1[name] = text + ("\x00" * (30 - len(text)))
+
+ if "COMM" in id3:
+ cmnt = id3["COMM"].text[0].encode('latin1', 'replace')[:28]
+ else: cmnt = ""
+ v1["comment"] = cmnt + ("\x00" * (29 - len(cmnt)))
+
+ if "TRCK" in id3:
+ try: v1["track"] = chr(+id3["TRCK"])
+ except ValueError: v1["track"] = "\x00"
+ else: v1["track"] = "\x00"
+
+ if "TCON" in id3:
+ try: genre = id3["TCON"].genres[0]
+ except IndexError: pass
+ else:
+ if genre in TCON.GENRES:
+ v1["genre"] = chr(TCON.GENRES.index(genre))
+ if "genre" not in v1: v1["genre"] = "\xff"
+
+ if "TDRC" in id3: v1["year"] = str(id3["TDRC"])[:4]
+ else: v1["year"] = "\x00\x00\x00\x00"
+
+ return ("TAG%(title)s%(artist)s%(album)s%(year)s%(comment)s"
+ "%(track)s%(genre)s") % v1
+
+class ID3FileType(mutagen.FileType):
+ """An unknown type of file with ID3 tags."""
+
+ class _Info(object):
+ length = 0
+ def __init__(self, fileobj, offset): pass
+ pprint = staticmethod(lambda: "Unknown format with ID3 tag")
+
+ def score(filename, fileobj, header):
+ return header.startswith("ID3")
+ score = staticmethod(score)
+
+ def add_tags(self, ID3=ID3):
+ """Add an empty ID3 tag to the file.
+
+ A custom tag reader may be used in instead of the default
+ mutagen.id3.ID3 object, e.g. an EasyID3 reader.
+ """
+ if self.tags is None:
+ self.tags = ID3()
+ else:
+ raise error("an ID3 tag already exists")
+
+ def load(self, filename, ID3=ID3, **kwargs):
+ """Load stream and tag information from a file.
+
+ A custom tag reader may be used in instead of the default
+ mutagen.id3.ID3 object, e.g. an EasyID3 reader.
+ """
+ self.filename = filename
+ try: self.tags = ID3(filename, **kwargs)
+ except error: self.tags = None
+ if self.tags is not None:
+ try: offset = self.tags.size
+ except AttributeError: offset = None
+ else: offset = None
+ try:
+ fileobj = file(filename, "rb")
+ self.info = self._Info(fileobj, offset)
+ finally:
+ fileobj.close()
+
--- /dev/null
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: m4a.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write MPEG-4 audio files with iTunes metadata.
+
+This module will read MPEG-4 audio information and metadata,
+as found in Apple's M4A (aka MP4, M4B, M4P) files.
+
+There is no official specification for this format. The source code
+for TagLib, FAAD, and various MPEG specifications at
+http://developer.apple.com/documentation/QuickTime/QTFF/,
+http://www.geocities.com/xhelmboyx/quicktime/formats/mp4-layout.txt,
+and http://wiki.multimedia.cx/index.php?title=Apple_QuickTime were all
+consulted.
+
+This module does not support 64 bit atom sizes, and so will not
+work on metadata over 4GB.
+"""
+
+import struct
+import sys
+
+from cStringIO import StringIO
+
+from mutagen import FileType, Metadata
+from mutagen._constants import GENRES
+from mutagen._util import cdata, insert_bytes, delete_bytes, DictProxy
+
+class error(IOError): pass
+class M4AMetadataError(error): pass
+class M4AStreamInfoError(error): pass
+class M4AMetadataValueError(ValueError, M4AMetadataError): pass
+
+import warnings
+warnings.warn(
+ "mutagen.m4a is deprecated; use mutagen.mp4 instead.", DeprecationWarning)
+
+# This is not an exhaustive list of container atoms, but just the
+# ones this module needs to peek inside.
+_CONTAINERS = ["moov", "udta", "trak", "mdia", "meta", "ilst",
+ "stbl", "minf", "stsd"]
+_SKIP_SIZE = { "meta": 4 }
+
+__all__ = ['M4A', 'Open', 'delete', 'M4ACover']
+
+class M4ACover(str):
+ """A cover artwork.
+
+ Attributes:
+ format -- format of the image (either FORMAT_JPEG or FORMAT_PNG)
+ """
+ FORMAT_JPEG = 0x0D
+ FORMAT_PNG = 0x0E
+
+ def __new__(cls, data, format=None):
+ self = str.__new__(cls, data)
+ if format is None: format= M4ACover.FORMAT_JPEG
+ self.format = format
+ return self
+
+class Atom(object):
+ """An individual atom.
+
+ Attributes:
+ children -- list child atoms (or None for non-container atoms)
+ length -- length of this atom, including length and name
+ name -- four byte name of the atom, as a str
+ offset -- location in the constructor-given fileobj of this atom
+
+ This structure should only be used internally by Mutagen.
+ """
+
+ children = None
+
+ def __init__(self, fileobj):
+ self.offset = fileobj.tell()
+ self.length, self.name = struct.unpack(">I4s", fileobj.read(8))
+ if self.length == 1:
+ raise error("64 bit atom sizes are not supported")
+ elif self.length < 8:
+ return
+
+ if self.name in _CONTAINERS:
+ self.children = []
+ fileobj.seek(_SKIP_SIZE.get(self.name, 0), 1)
+ while fileobj.tell() < self.offset + self.length:
+ self.children.append(Atom(fileobj))
+ else:
+ fileobj.seek(self.offset + self.length, 0)
+
+ def render(name, data):
+ """Render raw atom data."""
+ # this raises OverflowError if Py_ssize_t can't handle the atom data
+ size = len(data) + 8
+ if size <= 0xFFFFFFFF:
+ return struct.pack(">I4s", size, name) + data
+ else:
+ return struct.pack(">I4sQ", 1, name, size + 8) + data
+ render = staticmethod(render)
+
+ def __getitem__(self, remaining):
+ """Look up a child atom, potentially recursively.
+
+ e.g. atom['udta', 'meta'] => <Atom name='meta' ...>
+ """
+ if not remaining:
+ return self
+ elif self.children is None:
+ raise KeyError("%r is not a container" % self.name)
+ for child in self.children:
+ if child.name == remaining[0]:
+ return child[remaining[1:]]
+ else:
+ raise KeyError, "%r not found" % remaining[0]
+
+ def __repr__(self):
+ klass = self.__class__.__name__
+ if self.children is None:
+ return "<%s name=%r length=%r offset=%r>" % (
+ klass, self.name, self.length, self.offset)
+ else:
+ children = "\n".join([" " + line for child in self.children
+ for line in repr(child).splitlines()])
+ return "<%s name=%r length=%r offset=%r\n%s>" % (
+ klass, self.name, self.length, self.offset, children)
+
+class Atoms(object):
+ """Root atoms in a given file.
+
+ Attributes:
+ atoms -- a list of top-level atoms as Atom objects
+
+ This structure should only be used internally by Mutagen.
+ """
+ def __init__(self, fileobj):
+ self.atoms = []
+ fileobj.seek(0, 2)
+ end = fileobj.tell()
+ fileobj.seek(0)
+ while fileobj.tell() < end:
+ self.atoms.append(Atom(fileobj))
+
+ def path(self, *names):
+ """Look up and return the complete path of an atom.
+
+ For example, atoms.path('moov', 'udta', 'meta') will return a
+ list of three atoms, corresponding to the moov, udta, and meta
+ atoms.
+ """
+ path = [self]
+ for name in names:
+ path.append(path[-1][name,])
+ return path[1:]
+
+ def __getitem__(self, names):
+ """Look up a child atom.
+
+ 'names' may be a list of atoms (['moov', 'udta']) or a string
+ specifying the complete path ('moov.udta').
+ """
+ if isinstance(names, basestring):
+ names = names.split(".")
+ for child in self.atoms:
+ if child.name == names[0]:
+ return child[names[1:]]
+ else:
+ raise KeyError, "%s not found" % names[0]
+
+ def __repr__(self):
+ return "\n".join([repr(child) for child in self.atoms])
+
+class M4ATags(DictProxy, Metadata):
+ """Dictionary containing Apple iTunes metadata list key/values.
+
+ Keys are four byte identifiers, except for freeform ('----')
+ keys. Values are usually unicode strings, but some atoms have a
+ special structure:
+ cpil -- boolean
+ trkn, disk -- tuple of 16 bit ints (current, total)
+ tmpo -- 16 bit int
+ covr -- list of M4ACover objects (which are tagged strs)
+ gnre -- not supported. Use '\\xa9gen' instead.
+
+ The freeform '----' frames use a key in the format '----:mean:name'
+ where 'mean' is usually 'com.apple.iTunes' and 'name' is a unique
+ identifier for this frame. The value is a str, but is probably
+ text that can be decoded as UTF-8.
+
+ M4A tag data cannot exist outside of the structure of an M4A file,
+ so this class should not be manually instantiated.
+
+ Unknown non-text tags are removed.
+ """
+
+ def load(self, atoms, fileobj):
+ try: ilst = atoms["moov.udta.meta.ilst"]
+ except KeyError, key:
+ raise M4AMetadataError(key)
+ for atom in ilst.children:
+ fileobj.seek(atom.offset + 8)
+ data = fileobj.read(atom.length - 8)
+ parse = self.__atoms.get(atom.name, (M4ATags.__parse_text,))[0]
+ parse(self, atom, data)
+
+ def __key_sort((key1, v1), (key2, v2)):
+ # iTunes always writes the tags in order of "relevance", try
+ # to copy it as closely as possible.
+ order = ["\xa9nam", "\xa9ART", "\xa9wrt", "\xa9alb",
+ "\xa9gen", "gnre", "trkn", "disk",
+ "\xa9day", "cpil", "tmpo", "\xa9too",
+ "----", "covr", "\xa9lyr"]
+ order = dict(zip(order, range(len(order))))
+ last = len(order)
+ # If there's no key-based way to distinguish, order by length.
+ # If there's still no way, go by string comparison on the
+ # values, so we at least have something determinstic.
+ return (cmp(order.get(key1[:4], last), order.get(key2[:4], last)) or
+ cmp(len(v1), len(v2)) or cmp(v1, v2))
+ __key_sort = staticmethod(__key_sort)
+
+ def save(self, filename):
+ """Save the metadata to the given filename."""
+ values = []
+ items = self.items()
+ items.sort(self.__key_sort)
+ for key, value in items:
+ render = self.__atoms.get(
+ key[:4], (None, M4ATags.__render_text))[1]
+ values.append(render(self, key, value))
+ data = Atom.render("ilst", "".join(values))
+
+ # Find the old atoms.
+ fileobj = file(filename, "rb+")
+ try:
+ atoms = Atoms(fileobj)
+
+ moov = atoms["moov"]
+
+ if moov != atoms.atoms[-1]:
+ # "Free" the old moov block. Something in the mdat
+ # block is not happy when its offset changes and it
+ # won't play back. So, rather than try to figure that
+ # out, just move the moov atom to the end of the file.
+ offset = self.__move_moov(fileobj, moov)
+ else:
+ offset = 0
+
+ try:
+ path = atoms.path("moov", "udta", "meta", "ilst")
+ except KeyError:
+ self.__save_new(fileobj, atoms, data, offset)
+ else:
+ self.__save_existing(fileobj, atoms, path, data, offset)
+ finally:
+ fileobj.close()
+
+ def __move_moov(self, fileobj, moov):
+ fileobj.seek(moov.offset)
+ data = fileobj.read(moov.length)
+ fileobj.seek(moov.offset)
+ free = Atom.render("free", "\x00" * (moov.length - 8))
+ fileobj.write(free)
+ fileobj.seek(0, 2)
+ # Figure out how far we have to shift all our successive
+ # seek calls, relative to what the atoms say.
+ old_end = fileobj.tell()
+ fileobj.write(data)
+ return old_end - moov.offset
+
+ def __save_new(self, fileobj, atoms, ilst, offset):
+ hdlr = Atom.render("hdlr", "\x00" * 8 + "mdirappl" + "\x00" * 9)
+ meta = Atom.render("meta", "\x00\x00\x00\x00" + hdlr + ilst)
+ moov, udta = atoms.path("moov", "udta")
+ insert_bytes(fileobj, len(meta), udta.offset + offset + 8)
+ fileobj.seek(udta.offset + offset + 8)
+ fileobj.write(meta)
+ self.__update_parents(fileobj, [moov, udta], len(meta), offset)
+
+ def __save_existing(self, fileobj, atoms, path, data, offset):
+ # Replace the old ilst atom.
+ ilst = path.pop()
+ delta = len(data) - ilst.length
+ fileobj.seek(ilst.offset + offset)
+ if delta > 0:
+ insert_bytes(fileobj, delta, ilst.offset + offset)
+ elif delta < 0:
+ delete_bytes(fileobj, -delta, ilst.offset + offset)
+ fileobj.seek(ilst.offset + offset)
+ fileobj.write(data)
+ self.__update_parents(fileobj, path, delta, offset)
+
+ def __update_parents(self, fileobj, path, delta, offset):
+ # Update all parent atoms with the new size.
+ for atom in path:
+ fileobj.seek(atom.offset + offset)
+ size = cdata.uint_be(fileobj.read(4)) + delta
+ fileobj.seek(atom.offset + offset)
+ fileobj.write(cdata.to_uint_be(size))
+
+ def __render_data(self, key, flags, data):
+ data = struct.pack(">2I", flags, 0) + data
+ return Atom.render(key, Atom.render("data", data))
+
+ def __parse_freeform(self, atom, data):
+ try:
+ fileobj = StringIO(data)
+ mean_length = cdata.uint_be(fileobj.read(4))
+ # skip over 8 bytes of atom name, flags
+ mean = fileobj.read(mean_length - 4)[8:]
+ name_length = cdata.uint_be(fileobj.read(4))
+ name = fileobj.read(name_length - 4)[8:]
+ value_length = cdata.uint_be(fileobj.read(4))
+ # Name, flags, and reserved bytes
+ value = fileobj.read(value_length - 4)[12:]
+ except struct.error:
+ # Some ---- atoms have no data atom, I have no clue why
+ # they actually end up in the file.
+ pass
+ else:
+ self["%s:%s:%s" % (atom.name, mean, name)] = value
+ def __render_freeform(self, key, value):
+ dummy, mean, name = key.split(":", 2)
+ mean = struct.pack(">I4sI", len(mean) + 12, "mean", 0) + mean
+ name = struct.pack(">I4sI", len(name) + 12, "name", 0) + name
+ value = struct.pack(">I4s2I", len(value) + 16, "data", 0x1, 0) + value
+ final = mean + name + value
+ return Atom.render("----", mean + name + value)
+
+ def __parse_pair(self, atom, data):
+ self[atom.name] = struct.unpack(">2H", data[18:22])
+ def __render_pair(self, key, value):
+ track, total = value
+ if 0 <= track < 1 << 16 and 0 <= total < 1 << 16:
+ data = struct.pack(">4H", 0, track, total, 0)
+ return self.__render_data(key, 0, data)
+ else:
+ raise M4AMetadataValueError("invalid numeric pair %r" % (value,))
+
+ def __render_pair_no_trailing(self, key, value):
+ track, total = value
+ if 0 <= track < 1 << 16 and 0 <= total < 1 << 16:
+ data = struct.pack(">3H", 0, track, total)
+ return self.__render_data(key, 0, data)
+ else:
+ raise M4AMetadataValueError("invalid numeric pair %r" % (value,))
+
+ def __parse_genre(self, atom, data):
+ # Translate to a freeform genre.
+ genre = cdata.short_be(data[16:18])
+ if "\xa9gen" not in self:
+ try: self["\xa9gen"] = GENRES[genre - 1]
+ except IndexError: pass
+
+ def __parse_tempo(self, atom, data):
+ self[atom.name] = cdata.short_be(data[16:18])
+ def __render_tempo(self, key, value):
+ if 0 <= value < 1 << 16:
+ return self.__render_data(key, 0x15, cdata.to_ushort_be(value))
+ else:
+ raise M4AMetadataValueError("invalid short integer %r" % value)
+
+ def __parse_compilation(self, atom, data):
+ try: self[atom.name] = bool(ord(data[16:17]))
+ except TypeError: self[atom.name] = False
+
+ def __render_compilation(self, key, value):
+ return self.__render_data(key, 0x15, chr(bool(value)))
+
+ def __parse_cover(self, atom, data):
+ length, name, format = struct.unpack(">I4sI", data[:12])
+ if name != "data":
+ raise M4AMetadataError(
+ "unexpected atom %r inside 'covr'" % name)
+ if format not in (M4ACover.FORMAT_JPEG, M4ACover.FORMAT_PNG):
+ format = M4ACover.FORMAT_JPEG
+ self[atom.name]= M4ACover(data[16:length], format)
+ def __render_cover(self, key, value):
+ try: format = value.format
+ except AttributeError: format = M4ACover.FORMAT_JPEG
+ data = Atom.render("data", struct.pack(">2I", format, 0) + value)
+ return Atom.render(key, data)
+
+ def __parse_text(self, atom, data):
+ flags = cdata.uint_be(data[8:12])
+ if flags == 1:
+ self[atom.name] = data[16:].decode('utf-8', 'replace')
+ def __render_text(self, key, value):
+ return self.__render_data(key, 0x1, value.encode('utf-8'))
+
+ def delete(self, filename):
+ self.clear()
+ self.save(filename)
+
+ __atoms = {
+ "----": (__parse_freeform, __render_freeform),
+ "trkn": (__parse_pair, __render_pair),
+ "disk": (__parse_pair, __render_pair_no_trailing),
+ "gnre": (__parse_genre, None),
+ "tmpo": (__parse_tempo, __render_tempo),
+ "cpil": (__parse_compilation, __render_compilation),
+ "covr": (__parse_cover, __render_cover),
+ }
+
+ def pprint(self):
+ values = []
+ for key, value in self.iteritems():
+ key = key.decode('latin1')
+ try: values.append("%s=%s" % (key, value))
+ except UnicodeDecodeError:
+ values.append("%s=[%d bytes of data]" % (key, len(value)))
+ return "\n".join(values)
+
+class M4AInfo(object):
+ """MPEG-4 stream information.
+
+ Attributes:
+ bitrate -- bitrate in bits per second, as an int
+ length -- file length in seconds, as a float
+ """
+
+ bitrate = 0
+
+ def __init__(self, atoms, fileobj):
+ hdlr = atoms["moov.trak.mdia.hdlr"]
+ fileobj.seek(hdlr.offset)
+ if "soun" not in fileobj.read(hdlr.length):
+ raise M4AStreamInfoError("track has no audio data")
+
+ mdhd = atoms["moov.trak.mdia.mdhd"]
+ fileobj.seek(mdhd.offset)
+ data = fileobj.read(mdhd.length)
+ if ord(data[8]) == 0:
+ offset = 20
+ format = ">2I"
+ else:
+ offset = 28
+ format = ">IQ"
+ end = offset + struct.calcsize(format)
+ unit, length = struct.unpack(format, data[offset:end])
+ self.length = float(length) / unit
+
+ try:
+ atom = atoms["moov.trak.mdia.minf.stbl.stsd"]
+ fileobj.seek(atom.offset)
+ data = fileobj.read(atom.length)
+ self.bitrate = cdata.uint_be(data[-17:-13])
+ except (ValueError, KeyError):
+ # Bitrate values are optional.
+ pass
+
+ def pprint(self):
+ return "MPEG-4 audio, %.2f seconds, %d bps" % (
+ self.length, self.bitrate)
+
+class M4A(FileType):
+ """An MPEG-4 audio file, probably containing AAC.
+
+ If more than one track is present in the file, the first is used.
+ Only audio ('soun') tracks will be read.
+ """
+
+ _mimes = ["audio/mp4", "audio/x-m4a", "audio/mpeg4", "audio/aac"]
+
+ def load(self, filename):
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ atoms = Atoms(fileobj)
+ try: self.info = M4AInfo(atoms, fileobj)
+ except StandardError, err:
+ raise M4AStreamInfoError, err, sys.exc_info()[2]
+ try: self.tags = M4ATags(atoms, fileobj)
+ except M4AMetadataError:
+ self.tags = None
+ except StandardError, err:
+ raise M4AMetadataError, err, sys.exc_info()[2]
+ finally:
+ fileobj.close()
+
+ def add_tags(self):
+ self.tags = M4ATags()
+
+ def score(filename, fileobj, header):
+ return ("ftyp" in header) + ("mp4" in header)
+ score = staticmethod(score)
+
+Open = M4A
+
+def delete(filename):
+ """Remove tags from a file."""
+ M4A(filename).delete()
--- /dev/null
+# A Monkey's Audio (APE) reader/tagger
+#
+# Copyright 2006 Lukas Lalinsky <lalinsky@gmail.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: monkeysaudio.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Monkey's Audio streams with APEv2 tags.
+
+Monkey's Audio is a very efficient lossless audio compressor developed
+by Matt Ashland.
+
+For more information, see http://www.monkeysaudio.com/.
+"""
+
+__all__ = ["MonkeysAudio", "Open", "delete"]
+
+import struct
+
+from mutagen.apev2 import APEv2File, error, delete
+from mutagen._util import cdata
+
+class MonkeysAudioHeaderError(error): pass
+
+class MonkeysAudioInfo(object):
+ """Monkey's Audio stream information.
+
+ Attributes:
+ channels -- number of audio channels
+ length -- file length in seconds, as a float
+ sample_rate -- audio sampling rate in Hz
+ bits_per_sample -- bits per sample
+ version -- Monkey's Audio stream version, as a float (eg: 3.99)
+ """
+
+ def __init__(self, fileobj):
+ header = fileobj.read(76)
+ if len(header) != 76 or not header.startswith("MAC "):
+ raise MonkeysAudioHeaderError("not a Monkey's Audio file")
+ self.version = cdata.ushort_le(header[4:6])
+ if self.version >= 3980:
+ (blocks_per_frame, final_frame_blocks, total_frames,
+ self.bits_per_sample, self.channels,
+ self.sample_rate) = struct.unpack("<IIIHHI", header[56:76])
+ else:
+ compression_level = cdata.ushort_le(header[6:8])
+ self.channels, self.sample_rate = struct.unpack(
+ "<HI", header[10:16])
+ total_frames, final_frame_blocks = struct.unpack(
+ "<II", header[24:32])
+ if self.version >= 3950:
+ blocks_per_frame = 73728 * 4
+ elif self.version >= 3900 or (self.version >= 3800 and
+ compression_level == 4):
+ blocks_per_frame = 73728
+ else:
+ blocks_per_frame = 9216
+ self.version /= 1000.0
+ self.length = 0.0
+ if self.sample_rate != 0 and total_frames > 0:
+ total_blocks = ((total_frames - 1) * blocks_per_frame +
+ final_frame_blocks)
+ self.length = float(total_blocks) / self.sample_rate
+
+ def pprint(self):
+ return "Monkey's Audio %.2f, %.2f seconds, %d Hz" % (
+ self.version, self.length, self.sample_rate)
+
+class MonkeysAudio(APEv2File):
+ _Info = MonkeysAudioInfo
+ _mimes = ["audio/ape", "audio/x-ape"]
+
+ def score(filename, fileobj, header):
+ return header.startswith("MAC ") + filename.lower().endswith(".ape")
+ score = staticmethod(score)
+
+Open = MonkeysAudio
--- /dev/null
+# MP3 stream header information support for Mutagen.
+# Copyright 2006 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+
+"""MPEG audio stream information and tags."""
+
+import os
+import struct
+
+from mutagen.id3 import ID3FileType, BitPaddedInt, delete
+
+class error(RuntimeError): pass
+class HeaderNotFoundError(error, IOError): pass
+class InvalidMPEGHeader(error, IOError): pass
+
+# Mode values.
+STEREO, JOINTSTEREO, DUALCHANNEL, MONO = range(4)
+
+class MPEGInfo(object):
+ """MPEG audio stream information
+
+ Parse information about an MPEG audio file. This also reads the
+ Xing VBR header format.
+
+ This code was implemented based on the format documentation at
+ http://www.dv.co.yu/mpgscript/mpeghdr.htm.
+
+ Useful attributes:
+ length -- audio length, in seconds
+ bitrate -- audio bitrate, in bits per second
+ sketchy -- if true, the file may not be valid MPEG audio
+
+ Useless attributes:
+ version -- MPEG version (1, 2, 2.5)
+ layer -- 1, 2, or 3
+ mode -- One of STEREO, JOINTSTEREO, DUALCHANNEL, or MONO (0-3)
+ protected -- whether or not the file is "protected"
+ padding -- whether or not audio frames are padded
+ sample_rate -- audio sample rate, in Hz
+ """
+
+ # Map (version, layer) tuples to bitrates.
+ __BITRATE = {
+ (1, 1): range(0, 480, 32),
+ (1, 2): [0, 32, 48, 56, 64, 80, 96, 112,128,160,192,224,256,320,384],
+ (1, 3): [0, 32, 40, 48, 56, 64, 80, 96, 112,128,160,192,224,256,320],
+ (2, 1): [0, 32, 48, 56, 64, 80, 96, 112,128,144,160,176,192,224,256],
+ (2, 2): [0, 8, 16, 24, 32, 40, 48, 56, 64, 80, 96,112,128,144,160],
+ }
+
+ __BITRATE[(2, 3)] = __BITRATE[(2, 2)]
+ for i in range(1, 4): __BITRATE[(2.5, i)] = __BITRATE[(2, i)]
+
+ # Map version to sample rates.
+ __RATES = {
+ 1: [44100, 48000, 32000],
+ 2: [22050, 24000, 16000],
+ 2.5: [11025, 12000, 8000]
+ }
+
+ sketchy = False
+
+ def __init__(self, fileobj, offset=None):
+ """Parse MPEG stream information from a file-like object.
+
+ If an offset argument is given, it is used to start looking
+ for stream information and Xing headers; otherwise, ID3v2 tags
+ will be skipped automatically. A correct offset can make
+ loading files significantly faster.
+ """
+
+ try: size = os.path.getsize(fileobj.name)
+ except (IOError, OSError, AttributeError):
+ fileobj.seek(0, 2)
+ size = fileobj.tell()
+
+ # If we don't get an offset, try to skip an ID3v2 tag.
+ if offset is None:
+ fileobj.seek(0, 0)
+ idata = fileobj.read(10)
+ try: id3, insize = struct.unpack('>3sxxx4s', idata)
+ except struct.error: id3, insize = '', 0
+ insize = BitPaddedInt(insize)
+ if id3 == 'ID3' and insize > 0:
+ offset = insize
+ else: offset = 0
+
+ # Try to find two valid headers (meaning, very likely MPEG data)
+ # at the given offset, 30% through the file, 60% through the file,
+ # and 90% through the file.
+ for i in [offset, 0.3 * size, 0.6 * size, 0.9 * size]:
+ try: self.__try(fileobj, int(i), size - offset)
+ except error, e: pass
+ else: break
+ # If we can't find any two consecutive frames, try to find just
+ # one frame back at the original offset given.
+ else:
+ self.__try(fileobj, offset, size - offset, False)
+ self.sketchy = True
+
+ def __try(self, fileobj, offset, real_size, check_second=True):
+ # This is going to be one really long function; bear with it,
+ # because there's not really a sane point to cut it up.
+ fileobj.seek(offset, 0)
+
+ # We "know" we have an MPEG file if we find two frames that look like
+ # valid MPEG data. If we can't find them in 32k of reads, something
+ # is horribly wrong (the longest frame can only be about 4k). This
+ # is assuming the offset didn't lie.
+ data = fileobj.read(32768)
+
+ frame_1 = data.find("\xff")
+ while 0 <= frame_1 <= len(data) - 4:
+ frame_data = struct.unpack(">I", data[frame_1:frame_1 + 4])[0]
+ if (frame_data >> 16) & 0xE0 != 0xE0:
+ frame_1 = data.find("\xff", frame_1 + 2)
+ else:
+ version = (frame_data >> 19) & 0x3
+ layer = (frame_data >> 17) & 0x3
+ protection = (frame_data >> 16) & 0x1
+ bitrate = (frame_data >> 12) & 0xF
+ sample_rate = (frame_data >> 10) & 0x3
+ padding = (frame_data >> 9) & 0x1
+ private = (frame_data >> 8) & 0x1
+ self.mode = (frame_data >> 6) & 0x3
+ mode_extension = (frame_data >> 4) & 0x3
+ copyright = (frame_data >> 3) & 0x1
+ original = (frame_data >> 2) & 0x1
+ emphasis = (frame_data >> 0) & 0x3
+ if (version == 1 or layer == 0 or sample_rate == 0x3 or
+ bitrate == 0 or bitrate == 0xF):
+ frame_1 = data.find("\xff", frame_1 + 2)
+ else: break
+ else:
+ raise HeaderNotFoundError("can't sync to an MPEG frame")
+
+ # There is a serious problem here, which is that many flags
+ # in an MPEG header are backwards.
+ self.version = [2.5, None, 2, 1][version]
+ self.layer = 4 - layer
+ self.protected = not protection
+ self.padding = bool(padding)
+
+ self.bitrate = self.__BITRATE[(self.version, self.layer)][bitrate]
+ self.bitrate *= 1000
+ self.sample_rate = self.__RATES[self.version][sample_rate]
+
+ if self.layer == 1:
+ frame_length = (12 * self.bitrate / self.sample_rate + padding) * 4
+ frame_size = 384
+ else:
+ frame_length = 144 * self.bitrate / self.sample_rate + padding
+ frame_size = 1152
+
+ if check_second:
+ possible = frame_1 + frame_length
+ if possible > len(data) + 4:
+ raise HeaderNotFoundError("can't sync to second MPEG frame")
+ frame_data = struct.unpack(">H", data[possible:possible + 2])[0]
+ if frame_data & 0xFFE0 != 0xFFE0:
+ raise HeaderNotFoundError("can't sync to second MPEG frame")
+
+ frame_count = real_size / float(frame_length)
+ samples = frame_size * frame_count
+ self.length = samples / self.sample_rate
+
+ # Try to find/parse the Xing header, which trumps the above length
+ # and bitrate calculation.
+ fileobj.seek(offset, 0)
+ data = fileobj.read(32768)
+ try:
+ xing = data[:-4].index("Xing")
+ except ValueError:
+ # Try to find/parse the VBRI header, which trumps the above length
+ # calculation.
+ try:
+ vbri = data[:-24].index("VBRI")
+ except ValueError: pass
+ else:
+ # If a VBRI header was found, this is definitely MPEG audio.
+ self.sketchy = False
+ vbri_version = struct.unpack('>H', data[vbri + 4:vbri + 6])[0]
+ if vbri_version == 1:
+ frame_count = struct.unpack('>I', data[vbri + 14:vbri + 18])[0]
+ samples = frame_size * frame_count
+ self.length = (samples / self.sample_rate) or self.length
+ else:
+ # If a Xing header was found, this is definitely MPEG audio.
+ self.sketchy = False
+ flags = struct.unpack('>I', data[xing + 4:xing + 8])[0]
+ if flags & 0x1:
+ frame_count = struct.unpack('>I', data[xing + 8:xing + 12])[0]
+ samples = frame_size * frame_count
+ self.length = (samples / self.sample_rate) or self.length
+ if flags & 0x2:
+ bytes = struct.unpack('>I', data[xing + 12:xing + 16])[0]
+ self.bitrate = int((bytes * 8) // self.length)
+
+ def pprint(self):
+ s = "MPEG %s layer %d, %d bps, %s Hz, %.2f seconds" % (
+ self.version, self.layer, self.bitrate, self.sample_rate,
+ self.length)
+ if self.sketchy: s += " (sketchy)"
+ return s
+
+class MP3(ID3FileType):
+ """An MPEG audio (usually MPEG-1 Layer 3) file."""
+
+ _Info = MPEGInfo
+ _mimes = ["audio/mp3", "audio/x-mp3", "audio/mpeg", "audio/mpg",
+ "audio/x-mpeg"]
+
+ def score(filename, fileobj, header):
+ filename = filename.lower()
+ return (header.startswith("ID3") * 2 + filename.endswith(".mp3") +
+ filename.endswith(".mp2") + filename.endswith(".mpg") +
+ filename.endswith(".mpeg"))
+ score = staticmethod(score)
+
+Open = MP3
--- /dev/null
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: mp4.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write MPEG-4 audio files with iTunes metadata.
+
+This module will read MPEG-4 audio information and metadata,
+as found in Apple's MP4 (aka M4A, M4B, M4P) files.
+
+There is no official specification for this format. The source code
+for TagLib, FAAD, and various MPEG specifications at
+http://developer.apple.com/documentation/QuickTime/QTFF/,
+http://www.geocities.com/xhelmboyx/quicktime/formats/mp4-layout.txt,
+http://standards.iso.org/ittf/PubliclyAvailableStandards/c041828_ISO_IEC_14496-12_2005(E).zip,
+and http://wiki.multimedia.cx/index.php?title=Apple_QuickTime were all
+consulted.
+"""
+
+import struct
+import sys
+
+from mutagen import FileType, Metadata
+from mutagen._constants import GENRES
+from mutagen._util import cdata, insert_bytes, delete_bytes, DictProxy
+
+class error(IOError): pass
+class MP4MetadataError(error): pass
+class MP4StreamInfoError(error): pass
+class MP4MetadataValueError(ValueError, MP4MetadataError): pass
+
+# This is not an exhaustive list of container atoms, but just the
+# ones this module needs to peek inside.
+_CONTAINERS = ["moov", "udta", "trak", "mdia", "meta", "ilst",
+ "stbl", "minf", "moof", "traf"]
+_SKIP_SIZE = { "meta": 4 }
+
+__all__ = ['MP4', 'Open', 'delete', 'MP4Cover']
+
+class MP4Cover(str):
+ """A cover artwork.
+
+ Attributes:
+ format -- format of the image (either FORMAT_JPEG or FORMAT_PNG)
+ """
+ FORMAT_JPEG = 0x0D
+ FORMAT_PNG = 0x0E
+
+ def __new__(cls, data, format=None):
+ self = str.__new__(cls, data)
+ if format is None: format= MP4Cover.FORMAT_JPEG
+ self.format = format
+ return self
+
+class Atom(object):
+ """An individual atom.
+
+ Attributes:
+ children -- list child atoms (or None for non-container atoms)
+ length -- length of this atom, including length and name
+ name -- four byte name of the atom, as a str
+ offset -- location in the constructor-given fileobj of this atom
+
+ This structure should only be used internally by Mutagen.
+ """
+
+ children = None
+
+ def __init__(self, fileobj):
+ self.offset = fileobj.tell()
+ self.length, self.name = struct.unpack(">I4s", fileobj.read(8))
+ if self.length == 1:
+ self.length, = struct.unpack(">Q", fileobj.read(8))
+ elif self.length < 8:
+ return
+
+ if self.name in _CONTAINERS:
+ self.children = []
+ fileobj.seek(_SKIP_SIZE.get(self.name, 0), 1)
+ while fileobj.tell() < self.offset + self.length:
+ self.children.append(Atom(fileobj))
+ else:
+ fileobj.seek(self.offset + self.length, 0)
+
+ def render(name, data):
+ """Render raw atom data."""
+ # this raises OverflowError if Py_ssize_t can't handle the atom data
+ size = len(data) + 8
+ if size <= 0xFFFFFFFF:
+ return struct.pack(">I4s", size, name) + data
+ else:
+ return struct.pack(">I4sQ", 1, name, size + 8) + data
+ render = staticmethod(render)
+
+ def findall(self, name, recursive=False):
+ """Recursively find all child atoms by specified name."""
+ if self.children is not None:
+ for child in self.children:
+ if child.name == name:
+ yield child
+ if recursive:
+ for atom in child.findall(name, True):
+ yield atom
+
+ def __getitem__(self, remaining):
+ """Look up a child atom, potentially recursively.
+
+ e.g. atom['udta', 'meta'] => <Atom name='meta' ...>
+ """
+ if not remaining:
+ return self
+ elif self.children is None:
+ raise KeyError("%r is not a container" % self.name)
+ for child in self.children:
+ if child.name == remaining[0]:
+ return child[remaining[1:]]
+ else:
+ raise KeyError, "%r not found" % remaining[0]
+
+ def __repr__(self):
+ klass = self.__class__.__name__
+ if self.children is None:
+ return "<%s name=%r length=%r offset=%r>" % (
+ klass, self.name, self.length, self.offset)
+ else:
+ children = "\n".join([" " + line for child in self.children
+ for line in repr(child).splitlines()])
+ return "<%s name=%r length=%r offset=%r\n%s>" % (
+ klass, self.name, self.length, self.offset, children)
+
+class Atoms(object):
+ """Root atoms in a given file.
+
+ Attributes:
+ atoms -- a list of top-level atoms as Atom objects
+
+ This structure should only be used internally by Mutagen.
+ """
+ def __init__(self, fileobj):
+ self.atoms = []
+ fileobj.seek(0, 2)
+ end = fileobj.tell()
+ fileobj.seek(0)
+ while fileobj.tell() + 8 <= end:
+ self.atoms.append(Atom(fileobj))
+
+ def path(self, *names):
+ """Look up and return the complete path of an atom.
+
+ For example, atoms.path('moov', 'udta', 'meta') will return a
+ list of three atoms, corresponding to the moov, udta, and meta
+ atoms.
+ """
+ path = [self]
+ for name in names:
+ path.append(path[-1][name,])
+ return path[1:]
+
+ def __getitem__(self, names):
+ """Look up a child atom.
+
+ 'names' may be a list of atoms (['moov', 'udta']) or a string
+ specifying the complete path ('moov.udta').
+ """
+ if isinstance(names, basestring):
+ names = names.split(".")
+ for child in self.atoms:
+ if child.name == names[0]:
+ return child[names[1:]]
+ else:
+ raise KeyError, "%s not found" % names[0]
+
+ def __repr__(self):
+ return "\n".join([repr(child) for child in self.atoms])
+
+class MP4Tags(DictProxy, Metadata):
+ """Dictionary containing Apple iTunes metadata list key/values.
+
+ Keys are four byte identifiers, except for freeform ('----')
+ keys. Values are usually unicode strings, but some atoms have a
+ special structure:
+
+ Text values (multiple values per key are supported):
+ '\xa9nam' -- track title
+ '\xa9alb' -- album
+ '\xa9ART' -- artist
+ 'aART' -- album artist
+ '\xa9wrt' -- composer
+ '\xa9day' -- year
+ '\xa9cmt' -- comment
+ 'desc' -- description (usually used in podcasts)
+ 'purd' -- purchase date
+ '\xa9grp' -- grouping
+ '\xa9gen' -- genre
+ '\xa9lyr' -- lyrics
+ 'purl' -- podcast URL
+ 'egid' -- podcast episode GUID
+ 'catg' -- podcast category
+ 'keyw' -- podcast keywords
+ '\xa9too' -- encoded by
+ 'cprt' -- copyright
+ 'soal' -- album sort order
+ 'soaa' -- album artist sort order
+ 'soar' -- artist sort order
+ 'sonm' -- title sort order
+ 'soco' -- composer sort order
+ 'sosn' -- show sort order
+ 'tvsh' -- show name
+
+ Boolean values:
+ 'cpil' -- part of a compilation
+ 'pgap' -- part of a gapless album
+ 'pcst' -- podcast (iTunes reads this only on import)
+
+ Tuples of ints (multiple values per key are supported):
+ 'trkn' -- track number, total tracks
+ 'disk' -- disc number, total discs
+
+ Others:
+ 'tmpo' -- tempo/BPM, 16 bit int
+ 'covr' -- cover artwork, list of MP4Cover objects (which are
+ tagged strs)
+ 'gnre' -- ID3v1 genre. Not supported, use '\xa9gen' instead.
+
+ The freeform '----' frames use a key in the format '----:mean:name'
+ where 'mean' is usually 'com.apple.iTunes' and 'name' is a unique
+ identifier for this frame. The value is a str, but is probably
+ text that can be decoded as UTF-8. Multiple values per key are
+ supported.
+
+ MP4 tag data cannot exist outside of the structure of an MP4 file,
+ so this class should not be manually instantiated.
+
+ Unknown non-text tags are removed.
+ """
+
+ def load(self, atoms, fileobj):
+ try: ilst = atoms["moov.udta.meta.ilst"]
+ except KeyError, key:
+ raise MP4MetadataError(key)
+ for atom in ilst.children:
+ fileobj.seek(atom.offset + 8)
+ data = fileobj.read(atom.length - 8)
+ info = self.__atoms.get(atom.name, (MP4Tags.__parse_text, None))
+ info[0](self, atom, data, *info[2:])
+
+ def __key_sort((key1, v1), (key2, v2)):
+ # iTunes always writes the tags in order of "relevance", try
+ # to copy it as closely as possible.
+ order = ["\xa9nam", "\xa9ART", "\xa9wrt", "\xa9alb",
+ "\xa9gen", "gnre", "trkn", "disk",
+ "\xa9day", "cpil", "pgap", "pcst", "tmpo",
+ "\xa9too", "----", "covr", "\xa9lyr"]
+ order = dict(zip(order, range(len(order))))
+ last = len(order)
+ # If there's no key-based way to distinguish, order by length.
+ # If there's still no way, go by string comparison on the
+ # values, so we at least have something determinstic.
+ return (cmp(order.get(key1[:4], last), order.get(key2[:4], last)) or
+ cmp(len(v1), len(v2)) or cmp(v1, v2))
+ __key_sort = staticmethod(__key_sort)
+
+ def save(self, filename):
+ """Save the metadata to the given filename."""
+ values = []
+ items = self.items()
+ items.sort(self.__key_sort)
+ for key, value in items:
+ info = self.__atoms.get(key[:4], (None, MP4Tags.__render_text))
+ try:
+ values.append(info[1](self, key, value, *info[2:]))
+ except (TypeError, ValueError), s:
+ raise MP4MetadataValueError, s, sys.exc_info()[2]
+ data = Atom.render("ilst", "".join(values))
+
+ # Find the old atoms.
+ fileobj = file(filename, "rb+")
+ try:
+ atoms = Atoms(fileobj)
+ try:
+ path = atoms.path("moov", "udta", "meta", "ilst")
+ except KeyError:
+ self.__save_new(fileobj, atoms, data)
+ else:
+ self.__save_existing(fileobj, atoms, path, data)
+ finally:
+ fileobj.close()
+
+ def __pad_ilst(self, data, length=None):
+ if length is None:
+ length = ((len(data) + 1023) & ~1023) - len(data)
+ return Atom.render("free", "\x00" * length)
+
+ def __save_new(self, fileobj, atoms, ilst):
+ hdlr = Atom.render("hdlr", "\x00" * 8 + "mdirappl" + "\x00" * 9)
+ meta = Atom.render(
+ "meta", "\x00\x00\x00\x00" + hdlr + ilst + self.__pad_ilst(ilst))
+ try:
+ path = atoms.path("moov", "udta")
+ except KeyError:
+ # moov.udta not found -- create one
+ path = atoms.path("moov")
+ meta = Atom.render("udta", meta)
+ offset = path[-1].offset + 8
+ insert_bytes(fileobj, len(meta), offset)
+ fileobj.seek(offset)
+ fileobj.write(meta)
+ self.__update_parents(fileobj, path, len(meta))
+ self.__update_offsets(fileobj, atoms, len(meta), offset)
+
+ def __save_existing(self, fileobj, atoms, path, data):
+ # Replace the old ilst atom.
+ ilst = path.pop()
+ offset = ilst.offset
+ length = ilst.length
+
+ # Check for padding "free" atoms
+ meta = path[-1]
+ index = meta.children.index(ilst)
+ try:
+ prev = meta.children[index-1]
+ if prev.name == "free":
+ offset = prev.offset
+ length += prev.length
+ except IndexError:
+ pass
+ try:
+ next = meta.children[index+1]
+ if next.name == "free":
+ length += next.length
+ except IndexError:
+ pass
+
+ delta = len(data) - length
+ if delta > 0 or (delta < 0 and delta > -8):
+ data += self.__pad_ilst(data)
+ delta = len(data) - length
+ insert_bytes(fileobj, delta, offset)
+ elif delta < 0:
+ data += self.__pad_ilst(data, -delta - 8)
+ delta = 0
+
+ fileobj.seek(offset)
+ fileobj.write(data)
+ self.__update_parents(fileobj, path, delta)
+ self.__update_offsets(fileobj, atoms, delta, offset)
+
+ def __update_parents(self, fileobj, path, delta):
+ """Update all parent atoms with the new size."""
+ for atom in path:
+ fileobj.seek(atom.offset)
+ size = cdata.uint_be(fileobj.read(4)) + delta
+ fileobj.seek(atom.offset)
+ fileobj.write(cdata.to_uint_be(size))
+
+ def __update_offset_table(self, fileobj, fmt, atom, delta, offset):
+ """Update offset table in the specified atom."""
+ if atom.offset > offset:
+ atom.offset += delta
+ fileobj.seek(atom.offset + 12)
+ data = fileobj.read(atom.length - 12)
+ fmt = fmt % cdata.uint_be(data[:4])
+ offsets = struct.unpack(fmt, data[4:])
+ offsets = [o + (0, delta)[offset < o] for o in offsets]
+ fileobj.seek(atom.offset + 16)
+ fileobj.write(struct.pack(fmt, *offsets))
+
+ def __update_tfhd(self, fileobj, atom, delta, offset):
+ if atom.offset > offset:
+ atom.offset += delta
+ fileobj.seek(atom.offset + 9)
+ data = fileobj.read(atom.length - 9)
+ flags = cdata.uint_be("\x00" + data[:3])
+ if flags & 1:
+ o = cdata.ulonglong_be(data[7:15])
+ if o > offset:
+ o += delta
+ fileobj.seek(atom.offset + 16)
+ fileobj.write(cdata.to_ulonglong_be(o))
+
+ def __update_offsets(self, fileobj, atoms, delta, offset):
+ """Update offset tables in all 'stco' and 'co64' atoms."""
+ if delta == 0:
+ return
+ moov = atoms["moov"]
+ for atom in moov.findall('stco', True):
+ self.__update_offset_table(fileobj, ">%dI", atom, delta, offset)
+ for atom in moov.findall('co64', True):
+ self.__update_offset_table(fileobj, ">%dQ", atom, delta, offset)
+ try:
+ for atom in atoms["moof"].findall('tfhd', True):
+ self.__update_tfhd(fileobj, atom, delta, offset)
+ except KeyError:
+ pass
+
+ def __parse_data(self, atom, data):
+ pos = 0
+ while pos < atom.length - 8:
+ length, name, flags = struct.unpack(">I4sI", data[pos:pos+12])
+ if name != "data":
+ raise MP4MetadataError(
+ "unexpected atom %r inside %r" % (name, atom.name))
+ yield flags, data[pos+16:pos+length]
+ pos += length
+ def __render_data(self, key, flags, value):
+ return Atom.render(key, "".join([
+ Atom.render("data", struct.pack(">2I", flags, 0) + data)
+ for data in value]))
+
+ def __parse_freeform(self, atom, data):
+ length = cdata.uint_be(data[:4])
+ mean = data[12:length]
+ pos = length
+ length = cdata.uint_be(data[pos:pos+4])
+ name = data[pos+12:pos+length]
+ pos += length
+ value = []
+ while pos < atom.length - 8:
+ length, atom_name = struct.unpack(">I4s", data[pos:pos+8])
+ if atom_name != "data":
+ raise MP4MetadataError(
+ "unexpected atom %r inside %r" % (atom_name, atom.name))
+ value.append(data[pos+16:pos+length])
+ pos += length
+ if value:
+ self["%s:%s:%s" % (atom.name, mean, name)] = value
+ def __render_freeform(self, key, value):
+ dummy, mean, name = key.split(":", 2)
+ mean = struct.pack(">I4sI", len(mean) + 12, "mean", 0) + mean
+ name = struct.pack(">I4sI", len(name) + 12, "name", 0) + name
+ if isinstance(value, basestring):
+ value = [value]
+ return Atom.render("----", mean + name + "".join([
+ struct.pack(">I4s2I", len(data) + 16, "data", 1, 0) + data
+ for data in value]))
+
+ def __parse_pair(self, atom, data):
+ self[atom.name] = [struct.unpack(">2H", data[2:6]) for
+ flags, data in self.__parse_data(atom, data)]
+ def __render_pair(self, key, value):
+ data = []
+ for (track, total) in value:
+ if 0 <= track < 1 << 16 and 0 <= total < 1 << 16:
+ data.append(struct.pack(">4H", 0, track, total, 0))
+ else:
+ raise MP4MetadataValueError(
+ "invalid numeric pair %r" % ((track, total),))
+ return self.__render_data(key, 0, data)
+
+ def __render_pair_no_trailing(self, key, value):
+ data = []
+ for (track, total) in value:
+ if 0 <= track < 1 << 16 and 0 <= total < 1 << 16:
+ data.append(struct.pack(">3H", 0, track, total))
+ else:
+ raise MP4MetadataValueError(
+ "invalid numeric pair %r" % ((track, total),))
+ return self.__render_data(key, 0, data)
+
+ def __parse_genre(self, atom, data):
+ # Translate to a freeform genre.
+ genre = cdata.short_be(data[16:18])
+ if "\xa9gen" not in self:
+ try: self["\xa9gen"] = [GENRES[genre - 1]]
+ except IndexError: pass
+
+ def __parse_tempo(self, atom, data):
+ self[atom.name] = [cdata.ushort_be(value[1]) for
+ value in self.__parse_data(atom, data)]
+
+ def __render_tempo(self, key, value):
+ try:
+ if len(value) == 0:
+ return self.__render_data(key, 0x15, "")
+
+ if min(value) < 0 or max(value) >= 2**16:
+ raise MP4MetadataValueError(
+ "invalid 16 bit integers: %r" % value)
+ except TypeError:
+ raise MP4MetadataValueError(
+ "tmpo must be a list of 16 bit integers")
+
+ values = map(cdata.to_ushort_be, value)
+ return self.__render_data(key, 0x15, values)
+
+ def __parse_bool(self, atom, data):
+ try: self[atom.name] = bool(ord(data[16:17]))
+ except TypeError: self[atom.name] = False
+ def __render_bool(self, key, value):
+ return self.__render_data(key, 0x15, [chr(bool(value))])
+
+ def __parse_cover(self, atom, data):
+ self[atom.name] = []
+ pos = 0
+ while pos < atom.length - 8:
+ length, name, format = struct.unpack(">I4sI", data[pos:pos+12])
+ if name != "data":
+ raise MP4MetadataError(
+ "unexpected atom %r inside 'covr'" % name)
+ if format not in (MP4Cover.FORMAT_JPEG, MP4Cover.FORMAT_PNG):
+ format = MP4Cover.FORMAT_JPEG
+ cover = MP4Cover(data[pos+16:pos+length], format)
+ self[atom.name].append(MP4Cover(data[pos+16:pos+length], format))
+ pos += length
+ def __render_cover(self, key, value):
+ atom_data = []
+ for cover in value:
+ try: format = cover.format
+ except AttributeError: format = MP4Cover.FORMAT_JPEG
+ atom_data.append(
+ Atom.render("data", struct.pack(">2I", format, 0) + cover))
+ return Atom.render(key, "".join(atom_data))
+
+ def __parse_text(self, atom, data, expected_flags=1):
+ value = [text.decode('utf-8', 'replace') for flags, text
+ in self.__parse_data(atom, data)
+ if flags == expected_flags]
+ if value:
+ self[atom.name] = value
+ def __render_text(self, key, value, flags=1):
+ if isinstance(value, basestring):
+ value = [value]
+ return self.__render_data(
+ key, flags, [text.encode('utf-8') for text in value])
+
+ def delete(self, filename):
+ self.clear()
+ self.save(filename)
+
+ __atoms = {
+ "----": (__parse_freeform, __render_freeform),
+ "trkn": (__parse_pair, __render_pair),
+ "disk": (__parse_pair, __render_pair_no_trailing),
+ "gnre": (__parse_genre, None),
+ "tmpo": (__parse_tempo, __render_tempo),
+ "cpil": (__parse_bool, __render_bool),
+ "pgap": (__parse_bool, __render_bool),
+ "pcst": (__parse_bool, __render_bool),
+ "covr": (__parse_cover, __render_cover),
+ "purl": (__parse_text, __render_text, 0),
+ "egid": (__parse_text, __render_text, 0),
+ }
+
+ def pprint(self):
+ values = []
+ for key, value in self.iteritems():
+ key = key.decode('latin1')
+ if key == "covr":
+ values.append("%s=%s" % (key, ", ".join(
+ ["[%d bytes of data]" % len(data) for data in value])))
+ elif isinstance(value, list):
+ values.append("%s=%s" % (key, " / ".join(map(unicode, value))))
+ else:
+ values.append("%s=%s" % (key, value))
+ return "\n".join(values)
+
+class MP4Info(object):
+ """MPEG-4 stream information.
+
+ Attributes:
+ bitrate -- bitrate in bits per second, as an int
+ length -- file length in seconds, as a float
+ channels -- number of audio channels
+ sample_rate -- audio sampling rate in Hz
+ bits_per_sample -- bits per sample
+ """
+
+ bitrate = 0
+ channels = 0
+ sample_rate = 0
+ bits_per_sample = 0
+
+ def __init__(self, atoms, fileobj):
+ for trak in list(atoms["moov"].findall("trak")):
+ hdlr = trak["mdia", "hdlr"]
+ fileobj.seek(hdlr.offset)
+ data = fileobj.read(hdlr.length)
+ if data[16:20] == "soun":
+ break
+ else:
+ raise MP4StreamInfoError("track has no audio data")
+
+ mdhd = trak["mdia", "mdhd"]
+ fileobj.seek(mdhd.offset)
+ data = fileobj.read(mdhd.length)
+ if ord(data[8]) == 0:
+ offset = 20
+ format = ">2I"
+ else:
+ offset = 28
+ format = ">IQ"
+ end = offset + struct.calcsize(format)
+ unit, length = struct.unpack(format, data[offset:end])
+ self.length = float(length) / unit
+
+ try:
+ atom = trak["mdia", "minf", "stbl", "stsd"]
+ fileobj.seek(atom.offset)
+ data = fileobj.read(atom.length)
+ if data[20:24] == "mp4a":
+ length = cdata.uint_be(data[16:20])
+ (self.channels, self.bits_per_sample, _,
+ self.sample_rate) = struct.unpack(">3HI", data[40:50])
+ # ES descriptor type
+ if data[56:60] == "esds" and ord(data[64:65]) == 0x03:
+ pos = 65
+ # skip extended descriptor type tag, length, ES ID
+ # and stream priority
+ if data[pos:pos+3] == "\x80\x80\x80":
+ pos += 3
+ pos += 4
+ # decoder config descriptor type
+ if ord(data[pos]) == 0x04:
+ pos += 1
+ # skip extended descriptor type tag, length,
+ # object type ID, stream type, buffer size
+ # and maximum bitrate
+ if data[pos:pos+3] == "\x80\x80\x80":
+ pos += 3
+ pos += 10
+ # average bitrate
+ self.bitrate = cdata.uint_be(data[pos:pos+4])
+ except (ValueError, KeyError):
+ # stsd atoms are optional
+ pass
+
+ def pprint(self):
+ return "MPEG-4 audio, %.2f seconds, %d bps" % (
+ self.length, self.bitrate)
+
+class MP4(FileType):
+ """An MPEG-4 audio file, probably containing AAC.
+
+ If more than one track is present in the file, the first is used.
+ Only audio ('soun') tracks will be read.
+ """
+
+ _mimes = ["audio/mp4", "audio/x-m4a", "audio/mpeg4", "audio/aac"]
+
+ def load(self, filename):
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ atoms = Atoms(fileobj)
+ try: self.info = MP4Info(atoms, fileobj)
+ except StandardError, err:
+ raise MP4StreamInfoError, err, sys.exc_info()[2]
+ try: self.tags = MP4Tags(atoms, fileobj)
+ except MP4MetadataError:
+ self.tags = None
+ except StandardError, err:
+ raise MP4MetadataError, err, sys.exc_info()[2]
+ finally:
+ fileobj.close()
+
+ def add_tags(self):
+ self.tags = MP4Tags()
+
+ def score(filename, fileobj, header):
+ return ("ftyp" in header) + ("mp4" in header)
+ score = staticmethod(score)
+
+Open = MP4
+
+def delete(filename):
+ """Remove tags from a file."""
+ MP4(filename).delete()
--- /dev/null
+# A Musepack reader/tagger
+#
+# Copyright 2006 Lukas Lalinsky <lalinsky@gmail.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: musepack.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Musepack audio streams with APEv2 tags.
+
+Musepack is an audio format originally based on the MPEG-1 Layer-2
+algorithms. Stream versions 4 through 7 are supported.
+
+For more information, see http://www.musepack.net/.
+"""
+
+__all__ = ["Musepack", "Open", "delete"]
+
+import struct
+
+from mutagen.apev2 import APEv2File, error, delete
+from mutagen.id3 import BitPaddedInt
+from mutagen._util import cdata
+
+class MusepackHeaderError(error): pass
+
+RATES = [44100, 48000, 37800, 32000]
+
+class MusepackInfo(object):
+ """Musepack stream information.
+
+ Attributes:
+ channels -- number of audio channels
+ length -- file length in seconds, as a float
+ sample_rate -- audio sampling rate in Hz
+ bitrate -- audio bitrate, in bits per second
+ version -- Musepack stream version
+
+ Optional Attributes:
+ title_gain, title_peak -- Replay Gain and peak data for this song
+ album_gain, album_peak -- Replay Gain and peak data for this album
+
+ These attributes are only available in stream version 7. The
+ gains are a float, +/- some dB. The peaks are a percentage [0..1] of
+ the maximum amplitude. This means to get a number comparable to
+ VorbisGain, you must multiply the peak by 2.
+ """
+
+ def __init__(self, fileobj):
+ header = fileobj.read(32)
+ if len(header) != 32:
+ raise MusepackHeaderError("not a Musepack file")
+ # Skip ID3v2 tags
+ if header[:3] == "ID3":
+ size = 10 + BitPaddedInt(header[6:10])
+ fileobj.seek(size)
+ header = fileobj.read(32)
+ if len(header) != 32:
+ raise MusepackHeaderError("not a Musepack file")
+ # SV7
+ if header.startswith("MP+"):
+ self.version = ord(header[3]) & 0xF
+ if self.version < 7:
+ raise MusepackHeaderError("not a Musepack file")
+ frames = cdata.uint_le(header[4:8])
+ flags = cdata.uint_le(header[8:12])
+
+ self.title_peak, self.title_gain = struct.unpack(
+ "<Hh", header[12:16])
+ self.album_peak, self.album_gain = struct.unpack(
+ "<Hh", header[16:20])
+ self.title_gain /= 100.0
+ self.album_gain /= 100.0
+ self.title_peak /= 65535.0
+ self.album_peak /= 65535.0
+
+ self.sample_rate = RATES[(flags >> 16) & 0x0003]
+ self.bitrate = 0
+ # SV4-SV6
+ else:
+ header_dword = cdata.uint_le(header[0:4])
+ self.version = (header_dword >> 11) & 0x03FF;
+ if self.version < 4 or self.version > 6:
+ raise MusepackHeaderError("not a Musepack file")
+ self.bitrate = (header_dword >> 23) & 0x01FF;
+ self.sample_rate = 44100
+ if self.version >= 5:
+ frames = cdata.uint_le(header[4:8])
+ else:
+ frames = cdata.ushort_le(header[6:8])
+ if self.version < 6:
+ frames -= 1
+ self.channels = 2
+ self.length = float(frames * 1152 - 576) / self.sample_rate
+ if not self.bitrate and self.length != 0:
+ fileobj.seek(0, 2)
+ self.bitrate = int(fileobj.tell() * 8 / (self.length * 1000) + 0.5)
+
+ def pprint(self):
+ if self.version >= 7:
+ rg_data = ", Gain: %+0.2f (title), %+0.2f (album)" %(
+ self.title_gain, self.album_gain)
+ else:
+ rg_data = ""
+ return "Musepack, %.2f seconds, %d Hz%s" % (
+ self.length, self.sample_rate, rg_data)
+
+class Musepack(APEv2File):
+ _Info = MusepackInfo
+ _mimes = ["audio/x-musepack", "audio/x-mpc"]
+
+ def score(filename, fileobj, header):
+ return header.startswith("MP+") + filename.endswith(".mpc")
+ score = staticmethod(score)
+
+Open = Musepack
--- /dev/null
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: ogg.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write Ogg bitstreams and pages.
+
+This module reads and writes a subset of the Ogg bitstream format
+version 0. It does *not* read or write Ogg Vorbis files! For that,
+you should use mutagen.oggvorbis.
+
+This implementation is based on the RFC 3533 standard found at
+http://www.xiph.org/ogg/doc/rfc3533.txt.
+"""
+
+import struct
+import sys
+import zlib
+
+from cStringIO import StringIO
+
+from mutagen import FileType
+from mutagen._util import cdata, insert_bytes, delete_bytes
+
+class error(IOError):
+ """Ogg stream parsing errors."""
+ pass
+
+class OggPage(object):
+ """A single Ogg page (not necessarily a single encoded packet).
+
+ A page is a header of 26 bytes, followed by the length of the
+ data, followed by the data.
+
+ The constructor is givin a file-like object pointing to the start
+ of an Ogg page. After the constructor is finished it is pointing
+ to the start of the next page.
+
+ Attributes:
+ version -- stream structure version (currently always 0)
+ position -- absolute stream position (default -1)
+ serial -- logical stream serial number (default 0)
+ sequence -- page sequence number within logical stream (default 0)
+ offset -- offset this page was read from (default None)
+ complete -- if the last packet on this page is complete (default True)
+ packets -- list of raw packet data (default [])
+
+ Note that if 'complete' is false, the next page's 'continued'
+ property must be true (so set both when constructing pages).
+
+ If a file-like object is supplied to the constructor, the above
+ attributes will be filled in based on it.
+ """
+
+ version = 0
+ __type_flags = 0
+ position = 0L
+ serial = 0
+ sequence = 0
+ offset = None
+ complete = True
+
+ def __init__(self, fileobj=None):
+ self.packets = []
+
+ if fileobj is None:
+ return
+
+ self.offset = fileobj.tell()
+
+ header = fileobj.read(27)
+ if len(header) == 0:
+ raise EOFError
+
+ try:
+ (oggs, self.version, self.__type_flags, self.position,
+ self.serial, self.sequence, crc, segments) = struct.unpack(
+ "<4sBBqIIiB", header)
+ except struct.error:
+ raise error("unable to read full header; got %r" % header)
+
+ if oggs != "OggS":
+ raise error("read %r, expected %r, at 0x%x" % (
+ oggs, "OggS", fileobj.tell() - 27))
+
+ if self.version != 0:
+ raise error("version %r unsupported" % self.version)
+
+ total = 0
+ lacings = []
+ lacing_bytes = fileobj.read(segments)
+ if len(lacing_bytes) != segments:
+ raise error("unable to read %r lacing bytes" % segments)
+ for c in map(ord, lacing_bytes):
+ total += c
+ if c < 255:
+ lacings.append(total)
+ total = 0
+ if total:
+ lacings.append(total)
+ self.complete = False
+
+ self.packets = map(fileobj.read, lacings)
+ if map(len, self.packets) != lacings:
+ raise error("unable to read full data")
+
+ def __eq__(self, other):
+ """Two Ogg pages are the same if they write the same data."""
+ try:
+ return (self.write() == other.write())
+ except AttributeError:
+ return False
+
+ def __repr__(self):
+ attrs = ['version', 'position', 'serial', 'sequence', 'offset',
+ 'complete', 'continued', 'first', 'last']
+ values = ["%s=%r" % (attr, getattr(self, attr)) for attr in attrs]
+ return "<%s %s, %d bytes in %d packets>" % (
+ type(self).__name__, " ".join(values), sum(map(len, self.packets)),
+ len(self.packets))
+
+ def write(self):
+ """Return a string encoding of the page header and data.
+
+ A ValueError is raised if the data is too big to fit in a
+ single page.
+ """
+
+ data = [
+ struct.pack("<4sBBqIIi", "OggS", self.version, self.__type_flags,
+ self.position, self.serial, self.sequence, 0)
+ ]
+
+ lacing_data = []
+ for datum in self.packets:
+ quot, rem = divmod(len(datum), 255)
+ lacing_data.append("\xff" * quot + chr(rem))
+ lacing_data = "".join(lacing_data)
+ if not self.complete and lacing_data.endswith("\x00"):
+ lacing_data = lacing_data[:-1]
+ data.append(chr(len(lacing_data)))
+ data.append(lacing_data)
+ data.extend(self.packets)
+ data = "".join(data)
+
+ # Python's CRC is swapped relative to Ogg's needs.
+ crc = ~zlib.crc32(data.translate(cdata.bitswap), -1)
+ # Although we're using to_int_be, this actually makes the CRC
+ # a proper le integer, since Python's CRC is byteswapped.
+ crc = cdata.to_int_be(crc).translate(cdata.bitswap)
+ data = data[:22] + crc + data[26:]
+ return data
+
+ def __size(self):
+ size = 27 # Initial header size
+ for datum in self.packets:
+ quot, rem = divmod(len(datum), 255)
+ size += quot + 1
+ if not self.complete and rem == 0:
+ # Packet contains a multiple of 255 bytes and is not
+ # terminated, so we don't have a \x00 at the end.
+ size -= 1
+ size += sum(map(len, self.packets))
+ return size
+
+ size = property(__size, doc="Total frame size.")
+
+ def __set_flag(self, bit, val):
+ mask = 1 << bit
+ if val: self.__type_flags |= mask
+ else: self.__type_flags &= ~mask
+
+ continued = property(
+ lambda self: cdata.test_bit(self.__type_flags, 0),
+ lambda self, v: self.__set_flag(0, v),
+ doc="The first packet is continued from the previous page.")
+
+ first = property(
+ lambda self: cdata.test_bit(self.__type_flags, 1),
+ lambda self, v: self.__set_flag(1, v),
+ doc="This is the first page of a logical bitstream.")
+
+ last = property(
+ lambda self: cdata.test_bit(self.__type_flags, 2),
+ lambda self, v: self.__set_flag(2, v),
+ doc="This is the last page of a logical bitstream.")
+
+ def renumber(klass, fileobj, serial, start):
+ """Renumber pages belonging to a specified logical stream.
+
+ fileobj must be opened with mode r+b or w+b.
+
+ Starting at page number 'start', renumber all pages belonging
+ to logical stream 'serial'. Other pages will be ignored.
+
+ fileobj must point to the start of a valid Ogg page; any
+ occuring after it and part of the specified logical stream
+ will be numbered. No adjustment will be made to the data in
+ the pages nor the granule position; only the page number, and
+ so also the CRC.
+
+ If an error occurs (e.g. non-Ogg data is found), fileobj will
+ be left pointing to the place in the stream the error occured,
+ but the invalid data will be left intact (since this function
+ does not change the total file size).
+ """
+
+ number = start
+ while True:
+ try: page = OggPage(fileobj)
+ except EOFError:
+ break
+ else:
+ if page.serial != serial:
+ # Wrong stream, skip this page.
+ continue
+ # Changing the number can't change the page size,
+ # so seeking back based on the current size is safe.
+ fileobj.seek(-page.size, 1)
+ page.sequence = number
+ fileobj.write(page.write())
+ fileobj.seek(page.offset + page.size, 0)
+ number += 1
+ renumber = classmethod(renumber)
+
+ def to_packets(klass, pages, strict=False):
+ """Construct a list of packet data from a list of Ogg pages.
+
+ If strict is true, the first page must start a new packet,
+ and the last page must end the last packet.
+ """
+
+ serial = pages[0].serial
+ sequence = pages[0].sequence
+ packets = []
+
+ if strict:
+ if pages[0].continued:
+ raise ValueError("first packet is continued")
+ if not pages[-1].complete:
+ raise ValueError("last packet does not complete")
+ elif pages and pages[0].continued:
+ packets.append("")
+
+ for page in pages:
+ if serial != page.serial:
+ raise ValueError("invalid serial number in %r" % page)
+ elif sequence != page.sequence:
+ raise ValueError("bad sequence number in %r" % page)
+ else: sequence += 1
+
+ if page.continued: packets[-1] += page.packets[0]
+ else: packets.append(page.packets[0])
+ packets.extend(page.packets[1:])
+
+ return packets
+ to_packets = classmethod(to_packets)
+
+ def from_packets(klass, packets, sequence=0,
+ default_size=4096, wiggle_room=2048):
+ """Construct a list of Ogg pages from a list of packet data.
+
+ The algorithm will generate pages of approximately
+ default_size in size (rounded down to the nearest multiple of
+ 255). However, it will also allow pages to increase to
+ approximately default_size + wiggle_room if allowing the
+ wiggle room would finish a packet (only one packet will be
+ finished in this way per page; if the next packet would fit
+ into the wiggle room, it still starts on a new page).
+
+ This method reduces packet fragmentation when packet sizes are
+ slightly larger than the default page size, while still
+ ensuring most pages are of the average size.
+
+ Pages are numbered started at 'sequence'; other information is
+ uninitialized.
+ """
+
+ chunk_size = (default_size // 255) * 255
+
+ pages = []
+
+ page = OggPage()
+ page.sequence = sequence
+
+ for packet in packets:
+ page.packets.append("")
+ while packet:
+ data, packet = packet[:chunk_size], packet[chunk_size:]
+ if page.size < default_size and len(page.packets) < 255:
+ page.packets[-1] += data
+ else:
+ # If we've put any packet data into this page yet,
+ # we need to mark it incomplete. However, we can
+ # also have just started this packet on an already
+ # full page, in which case, just start the new
+ # page with this packet.
+ if page.packets[-1]:
+ page.complete = False
+ if len(page.packets) == 1:
+ page.position = -1L
+ else:
+ page.packets.pop(-1)
+ pages.append(page)
+ page = OggPage()
+ page.continued = not pages[-1].complete
+ page.sequence = pages[-1].sequence + 1
+ page.packets.append(data)
+
+ if len(packet) < wiggle_room:
+ page.packets[-1] += packet
+ packet = ""
+
+ if page.packets:
+ pages.append(page)
+
+ return pages
+ from_packets = classmethod(from_packets)
+
+ def replace(klass, fileobj, old_pages, new_pages):
+ """Replace old_pages with new_pages within fileobj.
+
+ old_pages must have come from reading fileobj originally.
+ new_pages are assumed to have the 'same' data as old_pages,
+ and so the serial and sequence numbers will be copied, as will
+ the flags for the first and last pages.
+
+ fileobj will be resized and pages renumbered as necessary. As
+ such, it must be opened r+b or w+b.
+ """
+
+ # Number the new pages starting from the first old page.
+ first = old_pages[0].sequence
+ for page, seq in zip(new_pages, range(first, first + len(new_pages))):
+ page.sequence = seq
+ page.serial = old_pages[0].serial
+
+ new_pages[0].first = old_pages[0].first
+ new_pages[0].last = old_pages[0].last
+ new_pages[0].continued = old_pages[0].continued
+
+ new_pages[-1].first = old_pages[-1].first
+ new_pages[-1].last = old_pages[-1].last
+ new_pages[-1].complete = old_pages[-1].complete
+ if not new_pages[-1].complete and len(new_pages[-1].packets) == 1:
+ new_pages[-1].position = -1L
+
+ new_data = "".join(map(klass.write, new_pages))
+
+ # Make room in the file for the new data.
+ delta = len(new_data)
+ fileobj.seek(old_pages[0].offset, 0)
+ insert_bytes(fileobj, delta, old_pages[0].offset)
+ fileobj.seek(old_pages[0].offset, 0)
+ fileobj.write(new_data)
+ new_data_end = old_pages[0].offset + delta
+
+ # Go through the old pages and delete them. Since we shifted
+ # the data down the file, we need to adjust their offsets. We
+ # also need to go backwards, so we don't adjust the deltas of
+ # the other pages.
+ old_pages.reverse()
+ for old_page in old_pages:
+ adj_offset = old_page.offset + delta
+ delete_bytes(fileobj, old_page.size, adj_offset)
+
+ # Finally, if there's any discrepency in length, we need to
+ # renumber the pages for the logical stream.
+ if len(old_pages) != len(new_pages):
+ fileobj.seek(new_data_end, 0)
+ serial = new_pages[-1].serial
+ sequence = new_pages[-1].sequence + 1
+ klass.renumber(fileobj, serial, sequence)
+ replace = classmethod(replace)
+
+ def find_last(klass, fileobj, serial):
+ """Find the last page of the stream 'serial'.
+
+ If the file is not multiplexed this function is fast. If it is,
+ it must read the whole the stream.
+
+ This finds the last page in the actual file object, or the last
+ page in the stream (with eos set), whichever comes first.
+ """
+
+ # For non-muxed streams, look at the last page.
+ try: fileobj.seek(-256*256, 2)
+ except IOError:
+ # The file is less than 64k in length.
+ fileobj.seek(0)
+ data = fileobj.read()
+ try: index = data.rindex("OggS")
+ except ValueError:
+ raise error("unable to find final Ogg header")
+ stringobj = StringIO(data[index:])
+ best_page = None
+ try:
+ page = OggPage(stringobj)
+ except error:
+ pass
+ else:
+ if page.serial == serial:
+ if page.last: return page
+ else: best_page = page
+ else: best_page = None
+
+ # The stream is muxed, so use the slow way.
+ fileobj.seek(0)
+ try:
+ page = OggPage(fileobj)
+ while not page.last:
+ page = OggPage(fileobj)
+ while page.serial != serial:
+ page = OggPage(fileobj)
+ best_page = page
+ return page
+ except error:
+ return best_page
+ except EOFError:
+ return best_page
+ find_last = classmethod(find_last)
+
+class OggFileType(FileType):
+ """An generic Ogg file."""
+
+ _Info = None
+ _Tags = None
+ _Error = None
+ _mimes = ["application/ogg", "application/x-ogg"]
+
+ def load(self, filename):
+ """Load file information from a filename."""
+
+ self.filename = filename
+ fileobj = file(filename, "rb")
+ try:
+ try:
+ self.info = self._Info(fileobj)
+ self.tags = self._Tags(fileobj, self.info)
+
+ if self.info.length:
+ # The streaminfo gave us real length information,
+ # don't waste time scanning the Ogg.
+ return
+
+ last_page = OggPage.find_last(fileobj, self.info.serial)
+ samples = last_page.position
+ try:
+ denom = self.info.sample_rate
+ except AttributeError:
+ denom = self.info.fps
+ self.info.length = samples / float(denom)
+
+ except error, e:
+ raise self._Error, e, sys.exc_info()[2]
+ except EOFError:
+ raise self._Error, "no appropriate stream found"
+ finally:
+ fileobj.close()
+
+ def delete(self, filename=None):
+ """Remove tags from a file.
+
+ If no filename is given, the one most recently loaded is used.
+ """
+ if filename is None:
+ filename = self.filename
+
+ self.tags.clear()
+ fileobj = file(filename, "rb+")
+ try:
+ try: self.tags._inject(fileobj)
+ except error, e:
+ raise self._Error, e, sys.exc_info()[2]
+ except EOFError:
+ raise self._Error, "no appropriate stream found"
+ finally:
+ fileobj.close()
+
+ def save(self, filename=None):
+ """Save a tag to a file.
+
+ If no filename is given, the one most recently loaded is used.
+ """
+ if filename is None:
+ filename = self.filename
+ fileobj = file(filename, "rb+")
+ try:
+ try: self.tags._inject(fileobj)
+ except error, e:
+ raise self._Error, e, sys.exc_info()[2]
+ except EOFError:
+ raise self._Error, "no appropriate stream found"
+ finally:
+ fileobj.close()
--- /dev/null
+# Ogg FLAC support.
+#
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: oggflac.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write Ogg FLAC comments.
+
+This module handles FLAC files wrapped in an Ogg bitstream. The first
+FLAC stream found is used. For 'naked' FLACs, see mutagen.flac.
+
+This module is bsaed off the specification at
+http://flac.sourceforge.net/ogg_mapping.html.
+"""
+
+__all__ = ["OggFLAC", "Open", "delete"]
+
+import struct
+
+from cStringIO import StringIO
+
+from mutagen.flac import StreamInfo, VCFLACDict
+from mutagen.ogg import OggPage, OggFileType, error as OggError
+
+class error(OggError): pass
+class OggFLACHeaderError(error): pass
+
+class OggFLACStreamInfo(StreamInfo):
+ """Ogg FLAC general header and stream info.
+
+ This encompasses the Ogg wrapper for the FLAC STREAMINFO metadata
+ block, as well as the Ogg codec setup that precedes it.
+
+ Attributes (in addition to StreamInfo's):
+ packets -- number of metadata packets
+ serial -- Ogg logical stream serial number
+ """
+
+ packets = 0
+ serial = 0
+
+ def load(self, data):
+ page = OggPage(data)
+ while not page.packets[0].startswith("\x7FFLAC"):
+ page = OggPage(data)
+ major, minor, self.packets, flac = struct.unpack(
+ ">BBH4s", page.packets[0][5:13])
+ if flac != "fLaC":
+ raise OggFLACHeaderError("invalid FLAC marker (%r)" % flac)
+ elif (major, minor) != (1, 0):
+ raise OggFLACHeaderError(
+ "unknown mapping version: %d.%d" % (major, minor))
+ self.serial = page.serial
+
+ # Skip over the block header.
+ stringobj = StringIO(page.packets[0][17:])
+ super(OggFLACStreamInfo, self).load(StringIO(page.packets[0][17:]))
+
+ def pprint(self):
+ return "Ogg " + super(OggFLACStreamInfo, self).pprint()
+
+class OggFLACVComment(VCFLACDict):
+ def load(self, data, info, errors='replace'):
+ # data should be pointing at the start of an Ogg page, after
+ # the first FLAC page.
+ pages = []
+ complete = False
+ while not complete:
+ page = OggPage(data)
+ if page.serial == info.serial:
+ pages.append(page)
+ complete = page.complete or (len(page.packets) > 1)
+ comment = StringIO(OggPage.to_packets(pages)[0][4:])
+ super(OggFLACVComment, self).load(comment, errors=errors)
+
+ def _inject(self, fileobj):
+ """Write tag data into the FLAC Vorbis comment packet/page."""
+
+ # Ogg FLAC has no convenient data marker like Vorbis, but the
+ # second packet - and second page - must be the comment data.
+ fileobj.seek(0)
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("\x7FFLAC"):
+ page = OggPage(fileobj)
+
+ first_page = page
+ while not (page.sequence == 1 and page.serial == first_page.serial):
+ page = OggPage(fileobj)
+
+ old_pages = [page]
+ while not (old_pages[-1].complete or len(old_pages[-1].packets) > 1):
+ page = OggPage(fileobj)
+ if page.serial == first_page.serial:
+ old_pages.append(page)
+
+ packets = OggPage.to_packets(old_pages, strict=False)
+
+ # Set the new comment block.
+ data = self.write()
+ data = packets[0][0] + struct.pack(">I", len(data))[-3:] + data
+ packets[0] = data
+
+ new_pages = OggPage.from_packets(packets, old_pages[0].sequence)
+ OggPage.replace(fileobj, old_pages, new_pages)
+
+class OggFLAC(OggFileType):
+ """An Ogg FLAC file."""
+
+ _Info = OggFLACStreamInfo
+ _Tags = OggFLACVComment
+ _Error = OggFLACHeaderError
+ _mimes = ["audio/x-oggflac"]
+
+ def score(filename, fileobj, header):
+ return (header.startswith("OggS") * (
+ ("FLAC" in header) + ("fLaC" in header)))
+ score = staticmethod(score)
+
+Open = OggFLAC
+
+def delete(filename):
+ """Remove tags from a file."""
+ OggFLAC(filename).delete()
--- /dev/null
+# Ogg Speex support.
+#
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: oggspeex.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write Ogg Speex comments.
+
+This module handles Speex files wrapped in an Ogg bitstream. The
+first Speex stream found is used.
+
+Read more about Ogg Speex at http://www.speex.org/. This module is
+based on the specification at http://www.speex.org/manual2/node7.html
+and clarifications after personal communication with Jean-Marc,
+http://lists.xiph.org/pipermail/speex-dev/2006-July/004676.html.
+"""
+
+__all__ = ["OggSpeex", "Open", "delete"]
+
+from mutagen._vorbis import VCommentDict
+from mutagen.ogg import OggPage, OggFileType, error as OggError
+from mutagen._util import cdata
+
+class error(OggError): pass
+class OggSpeexHeaderError(error): pass
+
+class OggSpeexInfo(object):
+ """Ogg Speex stream information.
+
+ Attributes:
+ bitrate - nominal bitrate in bits per second
+ channels - number of channels
+ length - file length in seconds, as a float
+
+ The reference encoder does not set the bitrate; in this case,
+ the bitrate will be 0.
+ """
+
+ length = 0
+
+ def __init__(self, fileobj):
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("Speex "):
+ page = OggPage(fileobj)
+ if not page.first:
+ raise OggSpeexHeaderError(
+ "page has ID header, but doesn't start a stream")
+ self.sample_rate = cdata.uint_le(page.packets[0][36:40])
+ self.channels = cdata.uint_le(page.packets[0][48:52])
+ self.bitrate = cdata.int_le(page.packets[0][52:56])
+ if self.bitrate == -1:
+ self.bitrate = 0
+ self.serial = page.serial
+
+ def pprint(self):
+ return "Ogg Speex, %.2f seconds" % self.length
+
+class OggSpeexVComment(VCommentDict):
+ """Speex comments embedded in an Ogg bitstream."""
+
+ def __init__(self, fileobj, info):
+ pages = []
+ complete = False
+ while not complete:
+ page = OggPage(fileobj)
+ if page.serial == info.serial:
+ pages.append(page)
+ complete = page.complete or (len(page.packets) > 1)
+ data = OggPage.to_packets(pages)[0] + "\x01"
+ super(OggSpeexVComment, self).__init__(data, framing=False)
+
+ def _inject(self, fileobj):
+ """Write tag data into the Speex comment packet/page."""
+
+ fileobj.seek(0)
+
+ # Find the first header page, with the stream info.
+ # Use it to get the serial number.
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("Speex "):
+ page = OggPage(fileobj)
+
+ # Look for the next page with that serial number, it'll start
+ # the comment packet.
+ serial = page.serial
+ page = OggPage(fileobj)
+ while page.serial != serial:
+ page = OggPage(fileobj)
+
+ # Then find all the pages with the comment packet.
+ old_pages = [page]
+ while not (old_pages[-1].complete or len(old_pages[-1].packets) > 1):
+ page = OggPage(fileobj)
+ if page.serial == old_pages[0].serial:
+ old_pages.append(page)
+
+ packets = OggPage.to_packets(old_pages, strict=False)
+
+ # Set the new comment packet.
+ packets[0] = self.write(framing=False)
+
+ new_pages = OggPage.from_packets(packets, old_pages[0].sequence)
+ OggPage.replace(fileobj, old_pages, new_pages)
+
+class OggSpeex(OggFileType):
+ """An Ogg Speex file."""
+
+ _Info = OggSpeexInfo
+ _Tags = OggSpeexVComment
+ _Error = OggSpeexHeaderError
+ _mimes = ["audio/x-speex"]
+
+ def score(filename, fileobj, header):
+ return (header.startswith("OggS") * ("Speex " in header))
+ score = staticmethod(score)
+
+Open = OggSpeex
+
+def delete(filename):
+ """Remove tags from a file."""
+ OggSpeex(filename).delete()
--- /dev/null
+# Ogg Theora support.
+#
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: oggtheora.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write Ogg Theora comments.
+
+This module handles Theora files wrapped in an Ogg bitstream. The
+first Theora stream found is used.
+
+Based on the specification at http://theora.org/doc/Theora_I_spec.pdf.
+"""
+
+__all__ = ["OggTheora", "Open", "delete"]
+
+import struct
+
+from mutagen._vorbis import VCommentDict
+from mutagen.ogg import OggPage, OggFileType, error as OggError
+
+class error(OggError): pass
+class OggTheoraHeaderError(error): pass
+
+class OggTheoraInfo(object):
+ """Ogg Theora stream information.
+
+ Attributes:
+ length - file length in seconds, as a float
+ fps - video frames per second, as a float
+ """
+
+ length = 0
+
+ def __init__(self, fileobj):
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("\x80theora"):
+ page = OggPage(fileobj)
+ if not page.first:
+ raise OggTheoraHeaderError(
+ "page has ID header, but doesn't start a stream")
+ data = page.packets[0]
+ vmaj, vmin = struct.unpack("2B", data[7:9])
+ if (vmaj, vmin) != (3, 2):
+ raise OggTheoraHeaderError(
+ "found Theora version %d.%d != 3.2" % (vmaj, vmin))
+ fps_num, fps_den = struct.unpack(">2I", data[22:30])
+ self.fps = fps_num / float(fps_den)
+ self.bitrate, = struct.unpack(">I", data[37:40] + "\x00")
+ self.serial = page.serial
+
+ def pprint(self):
+ return "Ogg Theora, %.2f seconds, %d bps" % (self.length, self.bitrate)
+
+class OggTheoraCommentDict(VCommentDict):
+ """Theora comments embedded in an Ogg bitstream."""
+
+ def __init__(self, fileobj, info):
+ pages = []
+ complete = False
+ while not complete:
+ page = OggPage(fileobj)
+ if page.serial == info.serial:
+ pages.append(page)
+ complete = page.complete or (len(page.packets) > 1)
+ data = OggPage.to_packets(pages)[0][7:]
+ super(OggTheoraCommentDict, self).__init__(data + "\x01")
+
+ def _inject(self, fileobj):
+ """Write tag data into the Theora comment packet/page."""
+
+ fileobj.seek(0)
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("\x81theora"):
+ page = OggPage(fileobj)
+
+ old_pages = [page]
+ while not (old_pages[-1].complete or len(old_pages[-1].packets) > 1):
+ page = OggPage(fileobj)
+ if page.serial == old_pages[0].serial:
+ old_pages.append(page)
+
+ packets = OggPage.to_packets(old_pages, strict=False)
+
+ packets[0] = "\x81theora" + self.write(framing=False)
+
+ new_pages = OggPage.from_packets(packets, old_pages[0].sequence)
+ OggPage.replace(fileobj, old_pages, new_pages)
+
+class OggTheora(OggFileType):
+ """An Ogg Theora file."""
+
+ _Info = OggTheoraInfo
+ _Tags = OggTheoraCommentDict
+ _Error = OggTheoraHeaderError
+ _mimes = ["video/x-theora"]
+
+ def score(filename, fileobj, header):
+ return (header.startswith("OggS") *
+ (("\x80theora" in header) + ("\x81theora" in header)))
+ score = staticmethod(score)
+
+Open = OggTheora
+
+def delete(filename):
+ """Remove tags from a file."""
+ OggTheora(filename).delete()
--- /dev/null
+# Ogg Vorbis support.
+#
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: oggvorbis.py 4275 2008-06-01 06:32:37Z piman $
+
+"""Read and write Ogg Vorbis comments.
+
+This module handles Vorbis files wrapped in an Ogg bitstream. The
+first Vorbis stream found is used.
+
+Read more about Ogg Vorbis at http://vorbis.com/. This module is based
+on the specification at http://www.xiph.org/vorbis/doc/Vorbis_I_spec.html.
+"""
+
+__all__ = ["OggVorbis", "Open", "delete"]
+
+import struct
+
+from mutagen._vorbis import VCommentDict
+from mutagen.ogg import OggPage, OggFileType, error as OggError
+
+class error(OggError): pass
+class OggVorbisHeaderError(error): pass
+
+class OggVorbisInfo(object):
+ """Ogg Vorbis stream information.
+
+ Attributes:
+ length - file length in seconds, as a float
+ bitrate - nominal ('average') bitrate in bits per second, as an int
+ """
+
+ length = 0
+
+ def __init__(self, fileobj):
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("\x01vorbis"):
+ page = OggPage(fileobj)
+ if not page.first:
+ raise OggVorbisHeaderError(
+ "page has ID header, but doesn't start a stream")
+ (self.channels, self.sample_rate, max_bitrate, nominal_bitrate,
+ min_bitrate) = struct.unpack("<B4I", page.packets[0][11:28])
+ self.serial = page.serial
+
+ if nominal_bitrate == 0:
+ self.bitrate = (max_bitrate + min_bitrate) // 2
+ elif max_bitrate and max_bitrate < nominal_bitrate:
+ # If the max bitrate is less than the nominal, we know
+ # the nominal is wrong.
+ self.bitrate = max_bitrate
+ elif min_bitrate > nominal_bitrate:
+ self.bitrate = min_bitrate
+ else:
+ self.bitrate = nominal_bitrate
+
+ def pprint(self):
+ return "Ogg Vorbis, %.2f seconds, %d bps" % (self.length, self.bitrate)
+
+class OggVCommentDict(VCommentDict):
+ """Vorbis comments embedded in an Ogg bitstream."""
+
+ def __init__(self, fileobj, info):
+ pages = []
+ complete = False
+ while not complete:
+ page = OggPage(fileobj)
+ if page.serial == info.serial:
+ pages.append(page)
+ complete = page.complete or (len(page.packets) > 1)
+ data = OggPage.to_packets(pages)[0][7:] # Strip off "\x03vorbis".
+ super(OggVCommentDict, self).__init__(data)
+
+ def _inject(self, fileobj):
+ """Write tag data into the Vorbis comment packet/page."""
+
+ # Find the old pages in the file; we'll need to remove them,
+ # plus grab any stray setup packet data out of them.
+ fileobj.seek(0)
+ page = OggPage(fileobj)
+ while not page.packets[0].startswith("\x03vorbis"):
+ page = OggPage(fileobj)
+
+ old_pages = [page]
+ while not (old_pages[-1].complete or len(old_pages[-1].packets) > 1):
+ page = OggPage(fileobj)
+ if page.serial == old_pages[0].serial:
+ old_pages.append(page)
+
+ packets = OggPage.to_packets(old_pages, strict=False)
+
+ # Set the new comment packet.
+ packets[0] = "\x03vorbis" + self.write()
+
+ new_pages = OggPage.from_packets(packets, old_pages[0].sequence)
+ OggPage.replace(fileobj, old_pages, new_pages)
+
+class OggVorbis(OggFileType):
+ """An Ogg Vorbis file."""
+
+ _Info = OggVorbisInfo
+ _Tags = OggVCommentDict
+ _Error = OggVorbisHeaderError
+ _mimes = ["audio/vorbis", "audio/x-vorbis"]
+
+ def score(filename, fileobj, header):
+ return (header.startswith("OggS") * ("\x01vorbis" in header))
+ score = staticmethod(score)
+
+Open = OggVorbis
+
+def delete(filename):
+ """Remove tags from a file."""
+ OggVorbis(filename).delete()
--- /dev/null
+# OptimFROG reader/tagger
+#
+# Copyright 2006 Lukas Lalinsky <lalinsky@gmail.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: optimfrog.py 4275 2008-06-01 06:32:37Z piman $
+
+"""OptimFROG audio streams with APEv2 tags.
+
+OptimFROG is a lossless audio compression program. Its main goal is to
+reduce at maximum the size of audio files, while permitting bit
+identical restoration for all input. It is similar with the ZIP
+compression, but it is highly specialized to compress audio data.
+
+Only versions 4.5 and higher are supported.
+
+For more information, see http://www.losslessaudio.org/
+"""
+
+__all__ = ["OptimFROG", "Open", "delete"]
+
+import struct
+from mutagen.apev2 import APEv2File, error, delete
+
+class OptimFROGHeaderError(error): pass
+
+class OptimFROGInfo(object):
+ """OptimFROG stream information.
+
+ Attributes:
+ channels - number of audio channels
+ length - file length in seconds, as a float
+ sample_rate - audio sampling rate in Hz
+ """
+
+ def __init__(self, fileobj):
+ header = fileobj.read(76)
+ if (len(header) != 76 or not header.startswith("OFR ") or
+ struct.unpack("<I", header[4:8])[0] not in [12, 15]):
+ raise OptimFROGHeaderError("not an OptimFROG file")
+ (total_samples, total_samples_high, sample_type, self.channels,
+ self.sample_rate) = struct.unpack("<IHBBI", header[8:20])
+ total_samples += total_samples_high << 32
+ self.channels += 1
+ if self.sample_rate:
+ self.length = float(total_samples) / (self.channels *
+ self.sample_rate)
+ else:
+ self.length = 0.0
+
+ def pprint(self):
+ return "OptimFROG, %.2f seconds, %d Hz" % (self.length,
+ self.sample_rate)
+
+class OptimFROG(APEv2File):
+ _Info = OptimFROGInfo
+
+ def score(filename, fileobj, header):
+ return (header.startswith("OFR") + filename.endswith(".ofr") +
+ filename.endswith(".ofs"))
+ score = staticmethod(score)
--- /dev/null
+# True Audio support for Mutagen
+# Copyright 2006 Joe Wreschnig
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of version 2 of the GNU General Public License as
+# published by the Free Software Foundation.
+
+"""True Audio audio stream information and tags.
+
+True Audio is a lossless format designed for real-time encoding and
+decoding. This module is based on the documentation at
+http://www.true-audio.com/TTA_Lossless_Audio_Codec_-_Format_Description
+
+True Audio files use ID3 tags.
+"""
+
+__all__ = ["TrueAudio", "Open", "delete"]
+
+from mutagen.id3 import ID3FileType, delete
+from mutagen._util import cdata
+
+class error(RuntimeError): pass
+class TrueAudioHeaderError(error, IOError): pass
+
+class TrueAudioInfo(object):
+ """True Audio stream information.
+
+ Attributes:
+ length - audio length, in seconds
+ sample_rate - audio sample rate, in Hz
+ """
+
+ def __init__(self, fileobj, offset):
+ fileobj.seek(offset or 0)
+ header = fileobj.read(18)
+ if len(header) != 18 or not header.startswith("TTA"):
+ raise TrueAudioHeaderError("TTA header not found")
+ self.sample_rate = cdata.int_le(header[10:14])
+ samples = cdata.uint_le(header[14:18])
+ self.length = float(samples) / self.sample_rate
+
+ def pprint(self):
+ return "True Audio, %.2f seconds, %d Hz." % (
+ self.length, self.sample_rate)
+
+class TrueAudio(ID3FileType):
+ """A True Audio file."""
+
+ _Info = TrueAudioInfo
+ _mimes = ["audio/x-tta"]
+
+ def score(filename, fileobj, header):
+ return (header.startswith("ID3") + header.startswith("TTA") +
+ filename.lower().endswith(".tta") * 2)
+ score = staticmethod(score)
+
+Open = TrueAudio
--- /dev/null
+# A WavPack reader/tagger
+#
+# Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# $Id: wavpack.py 4275 2008-06-01 06:32:37Z piman $
+
+"""WavPack reading and writing.
+
+WavPack is a lossless format that uses APEv2 tags. Read
+http://www.wavpack.com/ for more information.
+"""
+
+__all__ = ["WavPack", "Open", "delete"]
+
+from mutagen.apev2 import APEv2File, error, delete
+from mutagen._util import cdata
+
+class WavPackHeaderError(error): pass
+
+RATES = [6000, 8000, 9600, 11025, 12000, 16000, 22050, 24000, 32000, 44100,
+ 48000, 64000, 88200, 96000, 192000]
+
+class WavPackInfo(object):
+ """WavPack stream information.
+
+ Attributes:
+ channels - number of audio channels (1 or 2)
+ length - file length in seconds, as a float
+ sample_rate - audio sampling rate in Hz
+ version - WavPack stream version
+ """
+
+ def __init__(self, fileobj):
+ header = fileobj.read(28)
+ if len(header) != 28 or not header.startswith("wvpk"):
+ raise WavPackHeaderError("not a WavPack file")
+ samples = cdata.uint_le(header[12:16])
+ flags = cdata.uint_le(header[24:28])
+ self.version = cdata.short_le(header[8:10])
+ self.channels = bool(flags & 4) or 2
+ self.sample_rate = RATES[(flags >> 23) & 0xF]
+ self.length = float(samples) / self.sample_rate
+
+ def pprint(self):
+ return "WavPack, %.2f seconds, %d Hz" % (self.length, self.sample_rate)
+
+class WavPack(APEv2File):
+ _Info = WavPackInfo
+ _mimes = ["audio/x-wavpack"]
+
+ def score(filename, fileobj, header):
+ return header.startswith("wvpk") * 2
+ score = staticmethod(score)
fnp / wolnelektury.git / commitdiff