mutagen/_util.py

   1 # Copyright 2006 Joe Wreschnig <piman@sacredchao.net>
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License version 2 as
   5 # published by the Free Software Foundation.
   6 #
   7 # $Id: _util.py 4218 2007-12-02 06:11:20Z piman $
   8
   9 """Utility classes for Mutagen.
  10
  11 You should not rely on the interfaces here being stable. They are
  12 intended for internal use in Mutagen only.
  13 """
  14
  15 import struct
  16
  17 class DictMixin(object):
  18     """Implement the dict API using keys() and __*item__ methods.
  19
  20     Similar to UserDict.DictMixin, this takes a class that defines
  21     __getitem__, __setitem__, __delitem__, and keys(), and turns it
  22     into a full dict-like object.
  23
  24     UserDict.DictMixin is not suitable for this purpose because it's
  25     an old-style class.
  26
  27     This class is not optimized for very large dictionaries; many
  28     functions have linear memory requirements. I recommend you
  29     override some of these functions if speed is required.
  30     """
  31
  32     def __iter__(self):
  33         return iter(self.keys())
  34
  35     def has_key(self, key):
  36         try: self[key]
  37         except KeyError: return False
  38         else: return True
  39     __contains__ = has_key
  40
  41     iterkeys = lambda self: iter(self.keys())
  42
  43     def values(self):
  44         return map(self.__getitem__, self.keys())
  45     itervalues = lambda self: iter(self.values())
  46
  47     def items(self):
  48         return zip(self.keys(), self.values())
  49     iteritems = lambda s: iter(s.items())
  50
  51     def clear(self):
  52         map(self.__delitem__, self.keys())
  53
  54     def pop(self, key, *args):
  55         if len(args) > 1:
  56             raise TypeError("pop takes at most two arguments")
  57         try: value = self[key]
  58         except KeyError:
  59             if args: return args[0]
  60             else: raise
  61         del(self[key])
  62         return value
  63
  64     def popitem(self):
  65         try:
  66             key = self.keys()[0]
  67             return key, self.pop(key)
  68         except IndexError: raise KeyError("dictionary is empty")
  69
  70     def update(self, other=None, **kwargs):
  71         if other is None:
  72             self.update(kwargs)
  73             other = {}
  74
  75         try: map(self.__setitem__, other.keys(), other.values())
  76         except AttributeError:
  77             for key, value in other:
  78                 self[key] = value
  79
  80     def setdefault(self, key, default=None):
  81         try: return self[key]
  82         except KeyError:
  83             self[key] = default
  84             return default
  85
  86     def get(self, key, default=None):
  87         try: return self[key]
  88         except KeyError: return default
  89
  90     def __repr__(self):
  91         return repr(dict(self.items()))
  92
  93     def __cmp__(self, other):
  94         if other is None: return 1
  95         else: return cmp(dict(self.items()), other)
  96
  97     def __len__(self):
  98         return len(self.keys())
  99
 100 class DictProxy(DictMixin):
 101     def __init__(self, *args, **kwargs):
 102         self.__dict = {}
 103         super(DictProxy, self).__init__(*args, **kwargs)
 104
 105     def __getitem__(self, key):
 106         return self.__dict[key]
 107
 108     def __setitem__(self, key, value):
 109         self.__dict[key] = value
 110
 111     def __delitem__(self, key):
 112         del(self.__dict[key])
 113
 114     def keys(self):
 115         return self.__dict.keys()
 116
 117 class cdata(object):
 118     """C character buffer to Python numeric type conversions."""
 119
 120     from struct import error
 121
 122     short_le = staticmethod(lambda data: struct.unpack('<h', data)[0])
 123     ushort_le = staticmethod(lambda data: struct.unpack('<H', data)[0])
 124
 125     short_be = staticmethod(lambda data: struct.unpack('>h', data)[0])
 126     ushort_be = staticmethod(lambda data: struct.unpack('>H', data)[0])
 127
 128     int_le = staticmethod(lambda data: struct.unpack('<i', data)[0])
 129     uint_le = staticmethod(lambda data: struct.unpack('<I', data)[0])
 130
 131     int_be = staticmethod(lambda data: struct.unpack('>i', data)[0])
 132     uint_be = staticmethod(lambda data: struct.unpack('>I', data)[0])
 133
 134     longlong_le = staticmethod(lambda data: struct.unpack('<q', data)[0])
 135     ulonglong_le = staticmethod(lambda data: struct.unpack('<Q', data)[0])
 136
 137     longlong_be = staticmethod(lambda data: struct.unpack('>q', data)[0])
 138     ulonglong_be = staticmethod(lambda data: struct.unpack('>Q', data)[0])
 139
 140     to_short_le = staticmethod(lambda data: struct.pack('<h', data))
 141     to_ushort_le = staticmethod(lambda data: struct.pack('<H', data))
 142
 143     to_short_be = staticmethod(lambda data: struct.pack('>h', data))
 144     to_ushort_be = staticmethod(lambda data: struct.pack('>H', data))
 145
 146     to_int_le = staticmethod(lambda data: struct.pack('<i', data))
 147     to_uint_le = staticmethod(lambda data: struct.pack('<I', data))
 148
 149     to_int_be = staticmethod(lambda data: struct.pack('>i', data))
 150     to_uint_be = staticmethod(lambda data: struct.pack('>I', data))
 151
 152     to_longlong_le = staticmethod(lambda data: struct.pack('<q', data))
 153     to_ulonglong_le = staticmethod(lambda data: struct.pack('<Q', data))
 154
 155     to_longlong_be = staticmethod(lambda data: struct.pack('>q', data))
 156     to_ulonglong_be = staticmethod(lambda data: struct.pack('>Q', data))
 157
 158     bitswap = ''.join([chr(sum([((val >> i) & 1) << (7-i) for i in range(8)]))
 159                        for val in range(256)])
 160     del(i)
 161     del(val)
 162
 163     test_bit = staticmethod(lambda value, n: bool((value >> n) & 1))
 164
 165 def lock(fileobj):
 166     """Lock a file object 'safely'.
 167
 168     That means a failure to lock because the platform doesn't
 169     support fcntl or filesystem locks is not considered a
 170     failure. This call does block.
 171
 172     Returns whether or not the lock was successful, or
 173     raises an exception in more extreme circumstances (full
 174     lock table, invalid file).
 175     """
 176     try: import fcntl
 177     except ImportError:
 178         return False
 179     else:
 180         try: fcntl.lockf(fileobj, fcntl.LOCK_EX)
 181         except IOError:
 182             # FIXME: There's possibly a lot of complicated
 183             # logic that needs to go here in case the IOError
 184             # is EACCES or EAGAIN.
 185             return False
 186         else:
 187             return True
 188
 189 def unlock(fileobj):
 190     """Unlock a file object.
 191
 192     Don't call this on a file object unless a call to lock()
 193     returned true.
 194     """
 195     # If this fails there's a mismatched lock/unlock pair,
 196     # so we definitely don't want to ignore errors.
 197     import fcntl
 198     fcntl.lockf(fileobj, fcntl.LOCK_UN)
 199
 200 def insert_bytes(fobj, size, offset, BUFFER_SIZE=2**16):
 201     """Insert size bytes of empty space starting at offset.
 202
 203     fobj must be an open file object, open rb+ or
 204     equivalent. Mutagen tries to use mmap to resize the file, but
 205     falls back to a significantly slower method if mmap fails.
 206     """
 207     assert 0 < size
 208     assert 0 <= offset
 209     locked = False
 210     fobj.seek(0, 2)
 211     filesize = fobj.tell()
 212     movesize = filesize - offset
 213     fobj.write('\x00' * size)
 214     fobj.flush()
 215     try:
 216         try:
 217             import mmap
 218             map = mmap.mmap(fobj.fileno(), filesize + size)
 219             try: map.move(offset + size, offset, movesize)
 220             finally: map.close()
 221         except (ValueError, EnvironmentError, ImportError):
 222             # handle broken mmap scenarios
 223             locked = lock(fobj)
 224             fobj.truncate(filesize)
 225
 226             fobj.seek(0, 2)
 227             padsize = size
 228             # Don't generate an enormous string if we need to pad
 229             # the file out several megs.
 230             while padsize:
 231                 addsize = min(BUFFER_SIZE, padsize)
 232                 fobj.write("\x00" * addsize)
 233                 padsize -= addsize
 234
 235             fobj.seek(filesize, 0)
 236             while movesize:
 237                 # At the start of this loop, fobj is pointing at the end
 238                 # of the data we need to move, which is of movesize length.
 239                 thismove = min(BUFFER_SIZE, movesize)
 240                 # Seek back however much we're going to read this frame.
 241                 fobj.seek(-thismove, 1)
 242                 nextpos = fobj.tell()
 243                 # Read it, so we're back at the end.
 244                 data = fobj.read(thismove)
 245                 # Seek back to where we need to write it.
 246                 fobj.seek(-thismove + size, 1)
 247                 # Write it.
 248                 fobj.write(data)
 249                 # And seek back to the end of the unmoved data.
 250                 fobj.seek(nextpos)
 251                 movesize -= thismove
 252
 253             fobj.flush()
 254     finally:
 255         if locked:
 256             unlock(fobj)
 257
 258 def delete_bytes(fobj, size, offset, BUFFER_SIZE=2**16):
 259     """Delete size bytes of empty space starting at offset.
 260
 261     fobj must be an open file object, open rb+ or
 262     equivalent. Mutagen tries to use mmap to resize the file, but
 263     falls back to a significantly slower method if mmap fails.
 264     """
 265     locked = False
 266     assert 0 < size
 267     assert 0 <= offset
 268     fobj.seek(0, 2)
 269     filesize = fobj.tell()
 270     movesize = filesize - offset - size
 271     assert 0 <= movesize
 272     try:
 273         if movesize > 0:
 274             fobj.flush()
 275             try:
 276                 import mmap
 277                 map = mmap.mmap(fobj.fileno(), filesize)
 278                 try: map.move(offset, offset + size, movesize)
 279                 finally: map.close()
 280             except (ValueError, EnvironmentError, ImportError):
 281                 # handle broken mmap scenarios
 282                 locked = lock(fobj)
 283                 fobj.seek(offset + size)
 284                 buf = fobj.read(BUFFER_SIZE)
 285                 while buf:
 286                     fobj.seek(offset)
 287                     fobj.write(buf)
 288                     offset += len(buf)
 289                     fobj.seek(offset + size)
 290                     buf = fobj.read(BUFFER_SIZE)
 291         fobj.truncate(filesize - size)
 292         fobj.flush()
 293     finally:
 294         if locked:
 295             unlock(fobj)
 296
 297 def utf8(data):
 298     """Convert a basestring to a valid UTF-8 str."""
 299     if isinstance(data, str):
 300         return data.decode("utf-8", "replace").encode("utf-8")
 301     elif isinstance(data, unicode):
 302         return data.encode("utf-8")
 303     else: raise TypeError("only unicode/str types can be converted to UTF-8")