python-mutagen 1.31-1ubuntu1 / usr / lib / python2.7 / dist-packages / mutagen / asf / _objects.py

# -*- coding: utf-8 -*-
# Copyright (C) 2005-2006  Joe Wreschnig
# Copyright (C) 2006-2007  Lukas Lalinsky
#
# 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.

import struct

from mutagen._util import cdata, get_size
from mutagen._compat import text_type, xrange, izip
from mutagen._tags import PaddingInfo

from ._util import guid2bytes, bytes2guid, CODECS, ASFError, ASFHeaderError
from ._attrs import ASFBaseAttribute, ASFUnicodeAttribute


class BaseObject(object):
    """Base ASF object."""

    GUID = None
    _TYPES = {}

    def __init__(self):
        self.objects = []
        self.data = b""

    def parse(self, asf, data):
        self.data = data

    def render(self, asf):
        data = self.GUID + struct.pack("<Q", len(self.data) + 24) + self.data
        return data

    def get_child(self, guid):
        for obj in self.objects:
            if obj.GUID == guid:
                return obj
        return None

    @classmethod
    def _register(cls, other):
        cls._TYPES[other.GUID] = other
        return other

    @classmethod
    def _get_object(cls, guid):
        if guid in cls._TYPES:
            return cls._TYPES[guid]()
        else:
            return UnknownObject(guid)

    def __repr__(self):
        return "<%s GUID=%s objects=%r>" % (
            type(self).__name__, bytes2guid(self.GUID), self.objects)

    def pprint(self):
        l = []
        l.append("%s(%s)" % (type(self).__name__, bytes2guid(self.GUID)))
        for o in self.objects:
            for e in o.pprint().splitlines():
                l.append("  " + e)
        return "\n".join(l)


class UnknownObject(BaseObject):
    """Unknown ASF object."""

    def __init__(self, guid):
        super(UnknownObject, self).__init__()
        assert isinstance(guid, bytes)
        self.GUID = guid


@BaseObject._register
class HeaderObject(BaseObject):
    """ASF header."""

    GUID = guid2bytes("75B22630-668E-11CF-A6D9-00AA0062CE6C")

    @classmethod
    def parse_full(cls, asf, fileobj):
        """Raises ASFHeaderError"""

        header = cls()

        size, num_objects = cls.parse_size(fileobj)
        for i in xrange(num_objects):
            guid, size = struct.unpack("<16sQ", fileobj.read(24))
            obj = BaseObject._get_object(guid)
            data = fileobj.read(size - 24)
            obj.parse(asf, data)
            header.objects.append(obj)

        return header

    @classmethod
    def parse_size(cls, fileobj):
        """Returns (size, num_objects)

        Raises ASFHeaderError
        """

        header = fileobj.read(30)
        if len(header) != 30 or header[:16] != HeaderObject.GUID:
            raise ASFHeaderError("Not an ASF file.")

        return struct.unpack("<QL", header[16:28])

    def render_full(self, asf, fileobj, available, padding_func):
        # Render everything except padding
        num_objects = 0
        data = bytearray()
        for obj in self.objects:
            if obj.GUID == PaddingObject.GUID:
                continue
            data += obj.render(asf)
            num_objects += 1

        # calculate how much space we need at least
        padding_obj = PaddingObject()
        header_size = len(HeaderObject.GUID) + 14
        padding_overhead = len(padding_obj.render(asf))
        needed_size = len(data) + header_size + padding_overhead

        # ask the user for padding adjustments
        file_size = get_size(fileobj)
        content_size = file_size - available
        assert content_size >= 0
        info = PaddingInfo(available - needed_size, content_size)

        # add padding
        padding = info._get_padding(padding_func)
        padding_obj.parse(asf, b"\x00" * padding)
        data += padding_obj.render(asf)
        num_objects += 1

        data = (HeaderObject.GUID +
                struct.pack("<QL", len(data) + 30, num_objects) +
                b"\x01\x02" + data)

        return data

    def parse(self, asf, data):
        raise NotImplementedError

    def render(self, asf):
        raise NotImplementedError


@BaseObject._register
class ContentDescriptionObject(BaseObject):
    """Content description."""

    GUID = guid2bytes("75B22633-668E-11CF-A6D9-00AA0062CE6C")

    NAMES = [
        u"Title",
        u"Author",
        u"Copyright",
        u"Description",
        u"Rating",
    ]

    def parse(self, asf, data):
        super(ContentDescriptionObject, self).parse(asf, data)
        lengths = struct.unpack("<HHHHH", data[:10])
        texts = []
        pos = 10
        for length in lengths:
            end = pos + length
            if length > 0:
                texts.append(data[pos:end].decode("utf-16-le").strip(u"\x00"))
            else:
                texts.append(None)
            pos = end

        for key, value in izip(self.NAMES, texts):
            if value is not None:
                value = ASFUnicodeAttribute(value=value)
                asf._tags.setdefault(self.GUID, []).append((key, value))

    def render(self, asf):
        def render_text(name):
            value = asf.to_content_description.get(name)
            if value is not None:
                return text_type(value).encode("utf-16-le") + b"\x00\x00"
            else:
                return b""

        texts = [render_text(x) for x in self.NAMES]
        data = struct.pack("<HHHHH", *map(len, texts)) + b"".join(texts)
        return self.GUID + struct.pack("<Q", 24 + len(data)) + data


@BaseObject._register
class ExtendedContentDescriptionObject(BaseObject):
    """Extended content description."""

    GUID = guid2bytes("D2D0A440-E307-11D2-97F0-00A0C95EA850")

    def parse(self, asf, data):
        super(ExtendedContentDescriptionObject, self).parse(asf, data)
        num_attributes, = struct.unpack("<H", data[0:2])
        pos = 2
        for i in xrange(num_attributes):
            name_length, = struct.unpack("<H", data[pos:pos + 2])
            pos += 2
            name = data[pos:pos + name_length]
            name = name.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 = ASFBaseAttribute._get_type(value_type)(data=value)
            asf._tags.setdefault(self.GUID, []).append((name, attr))

    def render(self, asf):
        attrs = asf.to_extended_content_description.items()
        data = b"".join(attr.render(name) for (name, attr) in attrs)
        data = struct.pack("<QH", 26 + len(data), len(attrs)) + data
        return self.GUID + data


@BaseObject._register
class FilePropertiesObject(BaseObject):
    """File properties."""

    GUID = guid2bytes("8CABDCA1-A947-11CF-8EE4-00C00C205365")

    def parse(self, asf, data):
        super(FilePropertiesObject, self).parse(asf, data)
        length, _, preroll = struct.unpack("<QQQ", data[40:64])
        # there are files where preroll is larger than length, limit to >= 0
        asf.info.length = max((length / 10000000.0) - (preroll / 1000.0), 0.0)


@BaseObject._register
class StreamPropertiesObject(BaseObject):
    """Stream properties."""

    GUID = guid2bytes("B7DC0791-A9B7-11CF-8EE6-00C00C205365")

    def parse(self, asf, data):
        super(StreamPropertiesObject, self).parse(asf, data)
        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


@BaseObject._register
class CodecListObject(BaseObject):
    """Codec List"""

    GUID = guid2bytes("86D15240-311D-11D0-A3A4-00A0C90348F6")

    def _parse_entry(self, data, offset):
        """can raise cdata.error"""

        type_, offset = cdata.uint16_le_from(data, offset)

        units, offset = cdata.uint16_le_from(data, offset)
        # utf-16 code units, not characters..
        next_offset = offset + units * 2
        try:
            name = data[offset:next_offset].decode("utf-16-le").strip("\x00")
        except UnicodeDecodeError:
            name = u""
        offset = next_offset

        units, offset = cdata.uint16_le_from(data, offset)
        next_offset = offset + units * 2
        try:
            desc = data[offset:next_offset].decode("utf-16-le").strip("\x00")
        except UnicodeDecodeError:
            desc = u""
        offset = next_offset

        bytes_, offset = cdata.uint16_le_from(data, offset)
        next_offset = offset + bytes_
        codec = u""
        if bytes_ == 2:
            codec_id = cdata.uint16_le_from(data, offset)[0]
            if codec_id in CODECS:
                codec = CODECS[codec_id]
        offset = next_offset

        return offset, type_, name, desc, codec

    def parse(self, asf, data):
        super(CodecListObject, self).parse(asf, data)

        offset = 16
        count, offset = cdata.uint32_le_from(data, offset)
        for i in xrange(count):
            try:
                offset, type_, name, desc, codec = \
                    self._parse_entry(data, offset)
            except cdata.error:
                raise ASFError("invalid codec entry")

            # go with the first audio entry
            if type_ == 2:
                name = name.strip()
                desc = desc.strip()
                asf.info.codec_type = codec
                asf.info.codec_name = name
                asf.info.codec_description = desc
                return


@BaseObject._register
class PaddingObject(BaseObject):
    """Padding object"""

    GUID = guid2bytes("1806D474-CADF-4509-A4BA-9AABCB96AAE8")


@BaseObject._register
class StreamBitratePropertiesObject(BaseObject):
    """Stream bitrate properties"""

    GUID = guid2bytes("7BF875CE-468D-11D1-8D82-006097C9A2B2")


@BaseObject._register
class ContentEncryptionObject(BaseObject):
    """Content encryption"""

    GUID = guid2bytes("2211B3FB-BD23-11D2-B4B7-00A0C955FC6E")


@BaseObject._register
class ExtendedContentEncryptionObject(BaseObject):
    """Extended content encryption"""

    GUID = guid2bytes("298AE614-2622-4C17-B935-DAE07EE9289C")


@BaseObject._register
class HeaderExtensionObject(BaseObject):
    """Header extension."""

    GUID = guid2bytes("5FBF03B5-A92E-11CF-8EE3-00C00C205365")

    def parse(self, asf, data):
        super(HeaderExtensionObject, self).parse(asf, data)
        datasize, = struct.unpack("<I", data[18:22])
        datapos = 0
        while datapos < datasize:
            guid, size = struct.unpack(
                "<16sQ", data[22 + datapos:22 + datapos + 24])
            obj = BaseObject._get_object(guid)
            obj.parse(asf, data[22 + datapos + 24:22 + datapos + size])
            self.objects.append(obj)
            datapos += size

    def render(self, asf):
        data = bytearray()
        for obj in self.objects:
            # some files have the padding in the extension header, but we
            # want to add it at the end of the top level header. Just
            # skip padding at this level.
            if obj.GUID == PaddingObject.GUID:
                continue
            data += obj.render(asf)
        return (self.GUID + struct.pack("<Q", 24 + 16 + 6 + len(data)) +
                b"\x11\xD2\xD3\xAB\xBA\xA9\xcf\x11" +
                b"\x8E\xE6\x00\xC0\x0C\x20\x53\x65" +
                b"\x06\x00" + struct.pack("<I", len(data)) + data)


@BaseObject._register
class MetadataObject(BaseObject):
    """Metadata description."""

    GUID = guid2bytes("C5F8CBEA-5BAF-4877-8467-AA8C44FA4CCA")

    def parse(self, asf, data):
        super(MetadataObject, self).parse(asf, data)
        num_attributes, = struct.unpack("<H", data[0:2])
        pos = 2
        for i in xrange(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]
            name = name.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 = ASFBaseAttribute._get_type(value_type)(**args)
            asf._tags.setdefault(self.GUID, []).append((name, attr))

    def render(self, asf):
        attrs = asf.to_metadata.items()
        data = b"".join([attr.render_m(name) for (name, attr) in attrs])
        return (self.GUID + struct.pack("<QH", 26 + len(data), len(attrs)) +
                data)


@BaseObject._register
class MetadataLibraryObject(BaseObject):
    """Metadata library description."""

    GUID = guid2bytes("44231C94-9498-49D1-A141-1D134E457054")

    def parse(self, asf, data):
        super(MetadataLibraryObject, self).parse(asf, data)
        num_attributes, = struct.unpack("<H", data[0:2])
        pos = 2
        for i in xrange(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]
            name = name.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 = ASFBaseAttribute._get_type(value_type)(**args)
            asf._tags.setdefault(self.GUID, []).append((name, attr))

    def render(self, asf):
        attrs = asf.to_metadata_library
        data = b"".join([attr.render_ml(name) for (name, attr) in attrs])
        return (self.GUID + struct.pack("<QH", 26 + len(data), len(attrs)) +
                data)