ebml/core.py

   1 import datetime
   2 import struct
   3
   4
   5 __all__ = (
   6         'read_element_id',
   7         'read_element_size',
   8         'read_unsigned_integer',
   9         'read_signed_integer',
  10         'read_float',
  11         'read_string',
  12         'read_unicode_string',
  13         'read_date',
  14         'encode_element_id',
  15         'encode_element_size',
  16         'encode_unsigned_integer',
  17         'encode_signed_integer',
  18         'encode_float',
  19         'encode_string',
  20         'encode_unicode_string',
  21         'encode_date',
  22 )
  23
  24
  25 MAXIMUM_ELEMENT_ID_LENGTH = 4
  26 MAXIMUM_ELEMENT_SIZE_LENGTH = 8
  27 MAXIMUM_UNSIGNED_INTEGER_LENGTH = 8
  28 MAXIMUM_SIGNED_INTEGER_LENGTH = 8
  29
  30
  31 def maximum_element_size_for_length(length):
  32         """
  33
  34         Returns the maximum element size representable in a given number of bytes.
  35
  36         :arg length: the limit on the length of the encoded representation in bytes
  37         :type length: int
  38         :returns: the maximum element size representable
  39         :rtype: int
  40
  41         """
  42
  43         return (2**(7*length)) - 2
  44
  45
  46 def decode_vint_length(byte, mask=True):
  47         length = None
  48         value_mask = None
  49         for n in xrange(1, 9):
  50                 if byte & (2**8 - (2**(8 - n))) == 2**(8 - n):
  51                         length = n
  52                         value_mask = (2**(8 - n)) - 1
  53                         break
  54         if length is None:
  55                 raise IOError('Cannot decode invalid varible-length integer.')
  56         if mask:
  57                 byte = byte & value_mask
  58         return length, byte
  59
  60
  61 def read_element_id(stream):
  62         """
  63
  64         Reads an element ID from a file-like object.
  65
  66         :arg stream: the file-like object
  67         :returns: the decoded element ID and its length in bytes
  68         :rtype: tuple
  69
  70         """
  71
  72         byte = ord(stream.read(1))
  73         length, id_ = decode_vint_length(byte, False)
  74         if length > 4:
  75                 raise IOError('Cannot decode element ID with length > 8.')
  76         for i in xrange(0, length - 1):
  77                 byte = ord(stream.read(1))
  78                 id_ = (id_ * 2**8) + byte
  79         return id_, length
  80
  81
  82 def read_element_size(stream):
  83         """
  84
  85         Reads an element size from a file-like object.
  86
  87         :arg stream: the file-like object
  88         :returns: the decoded size (or None if unknown) and the length of the descriptor in bytes
  89         :rtype: tuple
  90
  91         """
  92
  93         byte = ord(stream.read(1))
  94         length, size = decode_vint_length(byte)
  95
  96         for i in xrange(0, length - 1):
  97                 byte = ord(stream.read(1))
  98                 size = (size * 2**8) + byte
  99
 100         if size == maximum_element_size_for_length(length) + 1:
 101                 size = None
 102
 103         return size, length
 104
 105
 106 def read_unsigned_integer(stream, size):
 107         """
 108
 109         Reads an encoded unsigned integer value from a file-like object.
 110
 111         :arg stream: the file-like object
 112         :arg size: the number of bytes to read and decode
 113         :type size: int
 114         :returns: the decoded unsigned integer value
 115         :rtype: int
 116
 117         """
 118
 119         value = 0
 120         for i in xrange(0, size):
 121                 byte = ord(stream.read(1))
 122                 value = (value << 8) | byte
 123         return value
 124
 125
 126 def read_signed_integer(stream, size):
 127         """
 128
 129         Reads an encoded signed integer value from a file-like object.
 130
 131         :arg stream: the file-like object
 132         :arg size: the number of bytes to read and decode
 133         :type size: int
 134         :returns: the decoded signed integer value
 135         :rtype: int
 136
 137         """
 138
 139         value = 0
 140         if size > 0:
 141                 byte = ord(stream.read(1))
 142                 if (byte & 0b10000000) == 0b10000000:
 143                         value = -1 << 8
 144                 value |= byte
 145                 for i in xrange(1, size):
 146                         byte = ord(stream.read(1))
 147                         value = (value << 1) | byte
 148         return value
 149
 150
 151 def read_float(stream, size):
 152         """
 153
 154         Reads an encoded floating point value from a file-like object.
 155
 156         :arg stream: the file-like object
 157         :arg size: the number of bytes to read and decode (must be 0, 4, or 8)
 158         :type size: int
 159         :returns: the decoded floating point value
 160         :rtype: float
 161
 162         """
 163
 164         if size not in (0, 4, 8):
 165                 raise IOError('Cannot read floating point values with lengths other than 0, 4, or 8 bytes.')
 166         value = 0.0
 167         if size in (4, 8):
 168                 data = stream.read(size)
 169                 value = struct.unpack({
 170                         4: '>f',
 171                         8: '>d'
 172                 }[size], data)[0]
 173         return value
 174
 175
 176 def read_string(stream, size):
 177         """
 178
 179         Reads an encoded ASCII string value from a file-like object.
 180
 181         :arg stream: the file-like object
 182         :arg size: the number of bytes to read and decode
 183         :type size: int
 184         :returns: the decoded ASCII string value
 185         :rtype: str
 186
 187         """
 188
 189         value = ''
 190         if size > 0:
 191                 value = stream.read(size)
 192         return value
 193
 194
 195 def read_unicode_string(stream, size):
 196         """
 197
 198         Reads an encoded unicode string value from a file-like object.
 199
 200         :arg stream: the file-like object
 201         :arg size: the number of bytes to read and decode
 202         :type size: int
 203         :returns: the decoded unicode string value
 204         :rtype: unicode
 205
 206         """
 207
 208         value = u''
 209         if size > 0:
 210                 data = stream.read(size)
 211                 value = unicode(data, 'utf_8')
 212         return value
 213
 214
 215 def read_date(stream, size):
 216         """
 217
 218         Reads an encoded date (and time) value from a file-like object.
 219
 220         :arg stream: the file-like object
 221         :arg size: the number of bytes to read and decode (must be 8)
 222         :type size: int
 223         :returns: the decoded date (and time) value
 224         :rtype: datetime
 225
 226         """
 227
 228         if size != 8:
 229                 raise IOError('Cannot read date values with lengths other than 8 bytes.')
 230         data = stream.read(size)
 231         nanoseconds = struct.unpack('>q', data)[0]
 232         delta = datetime.timedelta(microseconds=(nanoseconds // 1000))
 233         return datetime.datetime(2001, 1, 1, tzinfo=None) + delta
 234
 235
 236 def octet(n):
 237         """
 238
 239         Limits an integer or byte to 8 bits.
 240
 241         """
 242
 243         return n & 0b11111111
 244
 245
 246 def vint_mask_for_length(length):
 247         """
 248
 249         Returns the bitmask for the first byte of a variable-length integer (used for element ID and size descriptors).
 250
 251         :arg length: the length of the variable-length integer
 252         :type length: int
 253         :returns: the bitmask for the first byte of the variable-length integer
 254         :rtype: int
 255
 256         """
 257
 258         return 0b10000000 >> (length - 1)
 259
 260
 261 def encode_element_id(element_id):
 262         """
 263
 264         Encodes an element ID.
 265
 266         :arg element_id: an element ID
 267         :type element_id: int
 268         :returns: the encoded representation bytes
 269         :rtype: bytearray
 270
 271         """
 272
 273         length = MAXIMUM_ELEMENT_ID_LENGTH
 274         while length and not (element_id & (vint_mask_for_length(length) << ((length - 1) * 8))):
 275                 length -= 1
 276         if not length:
 277                 raise ValueError('Cannot encode invalid element ID %s.' % hex(element_id))
 278
 279         data = bytearray(length)
 280         for index in reversed(xrange(length)):
 281                 data[index] = octet(element_id)
 282                 element_id >>= 8
 283
 284         return data
 285
 286
 287 def encode_element_size(element_size, length=None):
 288         """
 289
 290         Encodes an element size. If element_size is None, the size will be encoded as unknown. If length is not None, the size will be encoded in that many bytes; otherwise, the size will be encoded in the minimum number of bytes required, or in 8 bytes if the size is unknown (element_size is None).
 291
 292         :arg element_size: the element size, or None if unknown
 293         :type element_size: int or None
 294         :arg length: the length of the encoded representation, or None for the minimum length required (defaults to None)
 295         :type length: int or None
 296         :returns: the encoded representation bytes
 297         :rtype: bytearray
 298
 299         """
 300
 301         if length is not None and (length < 1 or length > MAXIMUM_ELEMENT_SIZE_LENGTH):
 302                 raise ValueError('Cannot encode element sizes into representations shorter than one byte long or longer than %i bytes long.' % MAXIMUM_ELEMENT_SIZE_LENGTH)
 303         if element_size is not None:
 304                 if element_size > maximum_element_size_for_length(MAXIMUM_ELEMENT_SIZE_LENGTH if length is None else length):
 305                         raise ValueError('Cannot encode element size %i as it would have an encoded representation longer than %i bytes.' % (element_size, (MAXIMUM_ELEMENT_SIZE_LENGTH if length is None else length)))
 306                 req_length = 1
 307                 while (element_size >> ((req_length - 1) * 8)) >= (vint_mask_for_length(req_length) - 1) and req_length < MAXIMUM_ELEMENT_SIZE_LENGTH:
 308                         req_length += 1
 309                 if length is None:
 310                         length = req_length
 311         else:
 312                 if length is None:
 313                         length = 8 # other libraries do this, so unless another length is specified for the unknown size descriptor, do as they do to avoid compatibility issues.
 314                 element_size = maximum_element_size_for_length(length) + 1
 315
 316         data = bytearray(length)
 317         for index in reversed(xrange(length)):
 318                 data[index] = octet(element_size)
 319                 element_size >>= 8
 320                 if not index:
 321                         data[index] = data[index] | vint_mask_for_length(length)
 322
 323         return data
 324
 325
 326 def encode_unsigned_integer(uint, length=None):
 327         """
 328
 329         Encodes an unsigned integer value. If length is not None, uint will be encoded in that many bytes; otherwise, uint will be encoded in the minimum number of bytes required. If uint is None or 0, the minimum number of bytes required is 0.
 330
 331         :arg uint: the unsigned integer value
 332         :type uint: int
 333         :arg length: the length of the encoded representation, or None for the minimum length required (defaults to None)
 334         :type length: int or None
 335         :returns: the encoded representation bytes
 336         :rtype: bytearray
 337
 338         """
 339
 340         if uint is None:
 341                 uint = 0
 342         if uint > ((2**((MAXIMUM_UNSIGNED_INTEGER_LENGTH if length is None else length) * 8)) - 1):
 343                 raise ValueError('Cannot encode unsigned integer value %i as it would have an encoded representation longer than %i bytes.' % (uint, (MAXIMUM_UNSIGNED_INTEGER_LENGTH if length is None else length)))
 344         elif uint == 0:
 345                 req_length = 0
 346         else:
 347                 req_length = 1
 348                 while uint >= (1 << (req_length * 8)) and req_length < MAXIMUM_UNSIGNED_INTEGER_LENGTH:
 349                         req_length += 1
 350         if length is None:
 351                 length = req_length
 352
 353         data = bytearray(length)
 354         for index in reversed(xrange(length)):
 355                 data[index] = octet(uint)
 356                 uint >>= 8
 357
 358         return data
 359
 360
 361 def encode_signed_integer(sint, length=None):
 362         """
 363
 364         Encodes a signed integer value. If length is not None, sint will be encoded in that many bytes; otherwise, sint will be encoded in the minimum number of bytes required. If sint is None or 0, the minimum number of bytes required is 0.
 365
 366         :arg sint: the signed integer value
 367         :type sint: int
 368         :arg length: the length of the encoded representation, or None for the minimum length required (defaults to None)
 369         :type length: int or None
 370         :returns: the encoded representation bytes
 371         :rtype: bytearray
 372
 373         """
 374
 375         if sint is None:
 376                 sint = 0
 377         if not (-(2**(7+(8*((MAXIMUM_SIGNED_INTEGER_LENGTH if length is None else length)-1)))) <= sint <= (2**(7+(8*((MAXIMUM_SIGNED_INTEGER_LENGTH if length is None else length)-1))))-1):
 378                 raise ValueError('Cannot encode signed integer value %i as it would have an encoded representation longer than %i bytes.' % (sint, (MAXIMUM_SIGNED_INTEGER_LENGTH if length is None else length)))
 379         elif sint == 0:
 380                 req_length = 0
 381                 uint = 0
 382                 if length is None:
 383                         length = req_length
 384         else:
 385                 uint = ((-sint - 1) << 1) if sint < 0 else (sint << 1)
 386                 req_length = 1
 387                 while uint >= (1 << (req_length * 8)) and req_length < MAXIMUM_UNSIGNED_INTEGER_LENGTH:
 388                         req_length += 1
 389                 if length is None:
 390                         length = req_length
 391                 if sint >= 0:
 392                         uint = sint
 393                 else:
 394                         uint = 0b10000000 << (length - 1)
 395                         uint += sint
 396                         uint |= 0b10000000 << (length - 1)
 397
 398         data = bytearray(length)
 399         for index in reversed(xrange(length)):
 400                 data[index] = octet(uint)
 401                 uint >>= 8
 402
 403         return data
 404
 405
 406 def encode_float(float_, length=None):
 407         """
 408
 409         Encodes a floating point value. If length is not None, float_ will be encoded in that many bytes; otherwise, float_ will be encoded in 0 bytes if float_ is None or 0, and 8 bytes in all other cases. If float_ is not None or 0 and length is 0, ValueError will be raised.
 410
 411         :arg float_: the floating point value
 412         :type float_: float
 413         :arg length: the length of the encoded representation, or None (defaults to None)
 414         :type length: int or None
 415         :returns: the encoded representation bytes
 416         :rtype: bytearray
 417
 418         """
 419
 420         if length not in (None, 0, 4, 8):
 421                 raise ValueError('Cannot encode floating point values with lengths other than 0, 4, or 8 bytes.')
 422         if float_ is None:
 423                 float_ = 0.0
 424         if float_ == 0.0:
 425                 if length is None:
 426                         length = 0
 427         else:
 428                 if length is None:
 429                         length = 8
 430                 elif length == 0:
 431                         raise ValueError('Cannot encode floating point value %f as it would have an encoded representation longer than 0 bytes.' % float_)
 432
 433         if length in (4, 8):
 434                 data = bytearray(struct.pack({
 435                         4: '>f',
 436                         8: '>d'
 437                 }[length], float_))
 438         else:
 439                 data = bytearray()
 440
 441         return data
 442
 443
 444 def encode_string(string, length=None):
 445         """
 446
 447         Encodes an ASCII string value. If length is not None, string will be encoded in that many bytes by padding with zero bytes at the end if necessary; otherwise, string will be encoded in the minimum number of bytes required. If string is None or empty, the minimum number of bytes required is 0.
 448
 449         :arg string: the ASCII string value
 450         :type string: str
 451         :arg length: the length of the encoded representation, or None for the minimum length required (defaults to None)
 452         :type length: int or None
 453         :returns: the encoded representation bytes
 454         :rtype: bytearray
 455
 456         """
 457
 458         if string is None:
 459                 string = ''
 460         if length is None:
 461                 length = len(string)
 462         else:
 463                 if length < len(string):
 464                         raise ValueError('Cannot encode ASCII string value \'%s\' as it would have an encoded representation longer than %i bytes.' % (string, length))
 465                 elif length > len(string):
 466                         for i in xrange(0, (length - len(string))):
 467                                 string += chr(0)
 468
 469         return bytearray(string)
 470
 471
 472 def encode_unicode_string(string, length=None):
 473         """
 474
 475         Encodes a unicode string value. If length is not None, string will be encoded in that many bytes by padding with zero bytes at the end if necessary; otherwise, string will be encoded in the minimum number of bytes required. If string is None or empty, the minimum number of bytes required is 0.
 476
 477         :arg string: the unicode string value
 478         :type string: unicode
 479         :arg length: the length of the encoded representation, or None for the minimum length required (defaults to None)
 480         :type length: int or None
 481         :returns: the encoded representation bytes
 482         :rtype: bytearray
 483
 484         """
 485
 486         if string is None:
 487                 string = u''
 488         return encode_string(string.encode('utf_8'), length)
 489
 490
 491 def encode_date(date, length=None):
 492         """
 493
 494         Encodes a date (and time) value. If length is not None, it must be 8. If date is None, the current date (and time) will be encoded.
 495
 496         :arg date: the date (and time) value
 497         :type date: datetime.datettime
 498         :arg length: the length of the encoded representation (must be 8), or None
 499         :type length: int or None
 500         :returns: the encoded representation bytes
 501         :rtype: bytearray
 502
 503         """
 504
 505         if date is None:
 506                 date = datetime.datetime.utcnow()
 507         else:
 508                 date = (date - date.utcoffset()).replace(tzinfo=None)
 509         if length is None:
 510                 length = 8
 511         elif length != 8:
 512                 raise ValueError('Cannot encode date value %s with any length other than 8 bytes.')
 513
 514         delta = date - datetime.datetime(2001, 1, 1, tzinfo=None)
 515         nanoseconds = (delta.microseconds + ((delta.seconds + (delta.days * 24 * 60 * 60)) * 10**6)) * 10**3
 516         return encode_signed_integer(nanoseconds, length)