mkvinfo.py :  » Media-Sound-Audio » Media-Metadata-for-Python » mmpython-0.4.10 » video » Python Open Source

Home
Python Open Source
1.3.1.2 Python
2.Ajax
3.Aspect Oriented
4.Blog
5.Build
6.Business Application
7.Chart Report
8.Content Management Systems
9.Cryptographic
10.Database
11.Development
12.Editor
13.Email
14.ERP
15.Game 2D 3D
16.GIS
17.GUI
18.IDE
19.Installer
20.IRC
21.Issue Tracker
22.Language Interface
23.Log
24.Math
25.Media Sound Audio
26.Mobile
27.Network
28.Parser
29.PDF
30.Project Management
31.RSS
32.Search
33.Security
34.Template Engines
35.Test
36.UML
37.USB Serial
38.Web Frameworks
39.Web Server
40.Web Services
41.Web Unit
42.Wiki
43.Windows
44.XML
Python Open Source » Media Sound Audio » Media Metadata for Python 
Media Metadata for Python » mmpython 0.4.10 » video » mkvinfo.py
#if 0
# -----------------------------------------------------------------------
# mkvinfo.py - Matroska Streaming Video Files
# -----------------------------------------------------------------------
# $Id: mkvinfo.py 309 2004-04-18 17:55:26Z dischi $
#
# $Log$
# Revision 1.3  2004/04/18 17:55:26  dischi
# update, including subtitle support
#
# Revision 1.2  2004/03/21 08:57:31  dischi
# major bugfix
#
# Revision 1.1  2004/01/31 12:24:15  dischi
# add basic matroska info
#
# -----------------------------------------------------------------------
# MMPython - Media Metadata for Python
# Copyright (C) 2003 Thomas Schueppel, Dirk Meyer
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MER-
# CHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# -----------------------------------------------------------------------
#endif


from mmpython import mediainfo
import mmpython
import struct
import re
import stat
import os
import math
from types import *
from struct import *
from string import *

_print = mediainfo._debug

# Main IDs for the Matroska streams
MATROSKA_VIDEO_TRACK     = 0x01
MATROSKA_AUDIO_TRACK     = 0x02
MATROSKA_SUBTITLES_TRACK = 0x11

MATROSKA_HEADER_ID  = 0x1A45DFA3
MATROSKA_TRACKS_ID  = 0x1654AE6B
MATROSKA_SEGMENT_ID = 0x18538067
MATROSKA_SEGMENT_INFO_ID      = 0x1549A966
MATROSKA_CLUSTER_ID           = 0x1F43B675
MATROSKA_VOID_ID              = 0xEC
MATROSKA_CRC_ID               = 0xBF
MATROSKA_TIMECODESCALE_ID     = 0x2AD7B1
MATROSKA_DURATION_ID          = 0x4489
MATROSKA_CRC32_ID             = 0xBF
MATROSKA_TRACK_TYPE_ID        = 0x83
MATROSKA_TRACK_LANGUAGE_ID    = 0x22B59C
MATROSKA_TIMECODESCALE_ID     = 0x4489
MATROSKA_MUXING_APP_ID        = 0x4D80
MATROSKA_WRITING_APP_ID       = 0x5741
MATROSKA_CODEC_ID             = 0x86
MATROSKA_CODEC_NAME_ID        = 0x258688
MATROSKA_FRAME_DURATION_ID    = 0x23E383
MATROSKA_VIDEO_SETTINGS_ID    = 0xE0
MATROSKA_VID_WIDTH_ID         = 0xB0
MATROSKA_VID_HEIGHT_ID        = 0xBA
MATROSKA_AUDIO_SETTINGS_ID    = 0xE1
MATROSKA_AUDIO_SAMPLERATE_ID  = 0xB5
MATROSKA_AUDIO_CHANNELS_ID    = 0x9F
MATROSKA_TRACK_UID_ID         = 0x73C5
MATROSKA_TRACK_NUMBER_ID      = 0xD7

# This is class that is responsible to handle one Ebml entity as described in the Matroska/Ebml spec
class EbmlEntity:
    def __init__(self, inbuf):
        # Compute the EBML id
        # Set the CRC len to zero
        self.crc_len = 0
        # Now loop until we find an entity without CRC
        self.build_entity(inbuf)
        while self.get_id() == MATROSKA_CRC32_ID:
            self.crc_len += self.get_total_len()
            inbuf = inbuf[self.get_total_len():]
            self.build_entity(inbuf)

    def build_entity(self, inbuf):
        self.compute_id(inbuf)
        #_print("Entity id : %08X" % self.entity_id)
        if ( self.id_len == 0):
            self.valid = 0
            _print("EBML entity not found, bad file format")
            return
        self.valid = 1
        self.entity_len = self.compute_len(inbuf[self.id_len:])
        # Obviously, the segment can be very long (ie the whole file, so we truncate it at the read buffer size
        if (self.entity_len == -1):
            self.entity_data = inbuf[self.id_len+self.len_size:]
            self.entity_len = len(self.entity_data) # Set the remaining size
        else:
            self.entity_data = inbuf[self.id_len+self.len_size:self.id_len+self.len_size+self.entity_len]
        #_print("Entity len : %d" % self.entity_len)
        # if the size is 1, 2 3 or 4 it could be a numeric value, so do the job
        self.value = 0
        if self.entity_len == 1:
            self.value = ord(self.entity_data[0])
        if self.entity_len == 2:
            self.value = unpack('!H', self.entity_data)[0]
        if self.entity_len == 3:
            self.value = ord(self.entity_data[0])<<16 | ord(self.entity_data[1])<<8 | ord(self.entity_data[2])
        if self.entity_len == 4:
            self.value = unpack('!I', self.entity_data)[0]

    def compute_id(self, inbuf):
        first = ord(inbuf[0])
        self.id_len = 0
        if (first & 0x80):
            self.id_len = 1
            self.entity_id = first
        elif (first & 0x40):
            self.id_len = 2
            self.entity_id = ord(inbuf[0])<<8 | ord(inbuf[1])
        elif (first & 0x20):
            self.id_len = 3
            self.entity_id = (ord(inbuf[0])<<16) | (ord(inbuf[1])<<8) | (ord(inbuf[2]))
        elif (first & 0x10):
            self.id_len = 4
            self.entity_id = (ord(inbuf[0])<<24) | (ord(inbuf[1])<<16) | (ord(inbuf[2])<<8) | (ord(inbuf[3]))
        self.entity_str = inbuf[0:self.id_len]
        return

    def compute_len(self, inbuf):
        # Here we just handle the size up to 4 bytes
        # The size above will be truncated by the read buffer itself
        first = ord(inbuf[0])
        if (first & 0x80):
            self.len_size = 1
            return first - 0x80
        if (first & 0x40):
            self.len_size = 2
            (c1,c2) = unpack('BB',inbuf[:2])
            return ((c1-0x40)<<8) | (c2)
        if (first & 0x20):
            self.len_size = 3
            (c1, c2, c3) = unpack('BBB',inbuf[:3])
            return ((c1-0x20)<<16) | (c2<<8) | (c3)
        if (first & 0x10):
            self.len_size = 4
            (len) = unpack('!I',inbuf[:4])
            return len
        if (first & 0x08):
            self.len_size = 5
            return -1
        if (first & 0x04):
            self.len_size = 6
            return -1
        if (first & 0x02):
            self.len_size = 7
            return -1
        if (first & 0x01):
            self.len_size = 8
            return -1

    def get_crc_len(self):
        return self.crc_len

    def get_value(self):
        value = self.value
        return value

    def get_data(self):
        return self.entity_data

    def get_id(self):
        return self.entity_id

    def get_str_id(self):
        return self.entity_str

    def get_len(self):
        return self.entity_len

    def get_total_len(self):
        return self.entity_len+self.id_len+self.len_size


# This ithe main Matroska object
class MkvInfo(mediainfo.AVInfo):
    def __init__(self, file):
        mediainfo.AVInfo.__init__(self)
        self.samplerate = 1

        buffer = file.read(80000)
        if len(buffer) == 0:
            # Regular File end
            return None

        # Check the Matroska header
        header = EbmlEntity(buffer)
        if ( header.get_id() == MATROSKA_HEADER_ID ):
            _print("HEADER ID found %08X" % header.get_id() )
            self.valid = 1
            self.mime = 'application/mkv'
            self.type = 'Matroska'
            # Now get the segment
            segment = EbmlEntity(buffer[header.get_total_len():])
            if ( segment.get_id() == MATROSKA_SEGMENT_ID):
                _print("SEGMENT ID found %08X" % segment.get_id() )
                #MEDIACORE = ['title', 'caption', 'comment', 'artist', 'size', 'type', 'subtype',
                #'date', 'keywords', 'country', 'language', 'url']
                segtab = self.process_one_level(segment)
                seginfotab = self.process_one_level(segtab[MATROSKA_SEGMENT_INFO_ID])
                try:
                    # Express scalecode in ms instead of ns
                    # Rescale it to the second
                    scalecode = float(seginfotab[MATROSKA_TIMECODESCALE_ID].get_value() / (1000*1000))
                except:
                    scalecode = 1000
                try:
                    duration = float(unpack('!f', seginfotab[MATROSKA_DURATION_ID].get_data() )[0])
                    duration = float(duration / scalecode)
                    # Express the time in minutes
                    self.length = int(duration/60)
                except:
                    pass
                try:
                    _print ("Searching for id : %X" % MATROSKA_TRACKS_ID)
                    entity = segtab[MATROSKA_TRACKS_ID]
                    self.process_tracks(entity)
                except:
                    _print("TRACKS ID not found !!" )
            else:
                _print("SEGMENT ID not found %08X" % segment.get_id() )
        else:
            self.valid = 0

    def process_tracks(self, tracks):
        tracksbuf = tracks.get_data()
        indice = 0
        while indice < tracks.get_len():
            trackelem = EbmlEntity(tracksbuf[indice:])
            _print ("ELEMENT %X found" % trackelem.get_id())
            self.process_one_track(trackelem)
            indice += trackelem.get_total_len() + trackelem.get_crc_len()

    def process_one_level(self, item):
        buf = item.get_data()
        indice = 0
        tabelem = {}
        while indice < item.get_len():
            elem = EbmlEntity(buf[indice:])
            tabelem[elem.get_id()] = elem
            indice += elem.get_total_len() + elem.get_crc_len()
        return tabelem

    def process_one_track(self, track):
        # Process all the items at the track level
        tabelem = self.process_one_level(track)
        # We have the dict of track eleme, now build the MMPYTHON information
        type = tabelem[MATROSKA_TRACK_TYPE_ID]
        mytype = type.get_value()
        _print ("Track type found with UID %d" % mytype)
        if (mytype == MATROSKA_VIDEO_TRACK ):
            _print("VIDEO TRACK found !!" )
            #VIDEOCORE = ['length', 'encoder', 'bitrate', 'samplerate', 'codec', 'samplebits',
            #     'width', 'height', 'fps', 'aspect']
            vi = mediainfo.VideoInfo()
            try:
                elem = tabelem[MATROSKA_CODEC_ID]
                vi.codec = elem.get_data()
            except:
                vi.codec = 'Unknown'
            try:
                elem = tabelem[MATROSKA_FRAME_DURATION_ID]
                vi.fps = 1 / (pow(10, -9) * (elem.get_value()))
            except:
                vi.fps = 0
            try:
                vinfo = tabelem[MATROSKA_VIDEO_SETTINGS_ID]
                vidtab = self.process_one_level(vinfo)
                vi.width  = vidtab[MATROSKA_VID_WIDTH_ID].get_value()
                vi.height = vidtab[MATROSKA_VID_HEIGHT_ID].get_value()
            except:
                _print("No other info about video track !!!")
            self.video.append(vi)
        elif (mytype == MATROSKA_AUDIO_TRACK ):
            _print("AUDIO TRACK found !!" )
            #AUDIOCORE = ['channels', 'samplerate', 'length', 'encoder', 'codec', 'samplebits',
            #     'bitrate', 'language']
            ai = mediainfo.AudioInfo()
            try:
                elem = tabelem[MATROSKA_TRACK_LANGUAGE_ID]
                ai.language = elem.get_data()
                ai['language'] = elem.get_data()
            except:
                ai.language = 'en'
                ai['language'] = 'en'
            try:
                elem = tabelem[MATROSKA_CODEC_ID]
                ai.codec = elem.get_data()
            except:
                ai.codec = "Unknown"
            try:
                ainfo = tabelem[MATROSKA_AUDIO_SETTINGS_ID]
                audtab = self.process_one_level(vinfo)
                ai.samplerate  = unpack('!f', audtab[MATROSKA_AUDIO_SAMPLERATE_ID].get_value())[0]
                ai.channels = audtab[MATROSKA_AUDIO_CHANNELS_ID].get_value()
            except:
                _print("No other info about audio track !!!")
            self.audio.append(ai)
        elif (mytype == MATROSKA_SUBTITLES_TRACK):
            try:
                elem = tabelem[MATROSKA_TRACK_LANGUAGE_ID]
                language = elem.get_data()
                _print ("Subtitle language found : %s" % elem.get_data() )
            except:
                language = "en" # By default
            self.subtitles.append(language)

        #_print("Found %d elem for this track" % len(tabelem) )

mmpython.registertype( 'application/mkv', ('mkv', 'mka',), mediainfo.TYPE_AV, MkvInfo )
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.