Extracting FLV meta tags with Python

flv.py

from struct import unpack
from datetime import datetime


class FLVReader(dict):
    """
    Reads metadata from FLV files
    """

    # Tag types
    AUDIO = 8
    VIDEO = 9
    META = 18
    UNDEFINED = 0

    def __init__(self, filename):
        """
        Pass the filename of an flv file and it will return a dictionary of meta
        data.
        """
        # Lock on to the file
        self.file = open(filename, 'rb')
        self.signature = self.file.read(3)
        assert self.signature == 'FLV', 'Not an flv file'
        self.version = self.readbyte()
        self.typeFlags = self.readbyte()
        self.dataOffset = self.readint()
        extraDataLen = self.dataOffset - self.file.tell()
        self.extraData = self.file.read(extraDataLen)
        self.readtag()

    def readtag(self):
        self.readint()  # unknown
        tagType = self.readbyte()
        self.read24bit()  # dataSize
        self.read24bit()  # timeStamp
        self.readint()  # unknown
        if tagType == self.AUDIO:
            print "Can't handle audio tags yet"
        elif tagType == self.VIDEO:
            print "Can't handle video tags yet"
        elif tagType == self.META:
            # endpos = self.file.tell() + dataSize
            self.event = self.readAMFData()
            metaData = self.readAMFData()
            # We got the meta data.
            # Our job is done.
            # We are complete
            self.update(metaData)
        elif tagType == self.UNDEFINED:
            print "Can't handle undefined tags yet"

    def readint(self):
        data = self.file.read(4)
        return unpack('>I', data)[0]

    def readshort(self):
        data = self.file.read(2)
        return unpack('>H', data)[0]

    def readbyte(self):
        data = self.file.read(1)
        return unpack('B', data)[0]

    def read24bit(self):
        b1, b2, b3 = unpack('3B', self.file.read(3))
        return (b1 << 16) + (b2 << 8) + b3

    def readAMFData(self, dataType=None):
        if dataType is None:
            dataType = self.readbyte()
        funcs = {
            0: self.readAMFDouble,
            1: self.readAMFBoolean,
            2: self.readAMFString,
            3: self.readAMFObject,
            8: self.readAMFMixedArray,
            10: self.readAMFArray,
            11: self.readAMFDate
        }
        func = funcs[dataType]
        if callable(func):
            return func()

    def readAMFDouble(self):
        return unpack('>d', self.file.read(8))[0]

    def readAMFBoolean(self):
        return self.readbyte() == 1

    def readAMFString(self):
        size = self.readshort()
        return self.file.read(size)

    def readAMFObject(self):
        result = {}
        data = True
        while data:
            size = self.readshort()
            key = self.file.read(size)
            dataType = self.readbyte()
            if not key and dataType == 9:
                break
            data = self.readAMFData(dataType)
            result[key] = data
        return result

    def readAMFMixedArray(self):
        size = self.readint()
        result = {}
        i = 0
        while i < size:
            key = self.readAMFString()
            dataType = self.readbyte()
            if not key and dataType == 9:
                break
            result[key] = self.readAMFData(dataType)
            i += 1
        return result

    def readAMFArray(self):
        size = self.readint()
        result = []
        i = 0
        while i < size:
            result.append(self.readAMFData())
            i += 1
        return result

    def readAMFDate(self):
        date = self.readAMFDouble() / 1000
        self.readshort()  # localoffset
        return datetime.fromtimestamp(date)

if __name__ == '__main__':
    import sys
    from pprint import pprint
    if len(sys.argv) == 1:
        print 'Usage: %s filename [filename]...' % sys.argv[0]
        print 'Where filename is a .flv file'
        print 'eg. %s myfile.flv' % sys.argv[0]
    for fn in sys.argv[1:]:
        x = FLVReader(fn)
        pprint(x)

Hey,

I'm trying to convert audio data with FLV tags to PCM or WAV format.

I used Python's library librtmp to read audio data from server using the following code:

import librtmp

# Create a connection
conn = librtmp.RTMP("rtmp://your.server.net/app/playpath", live=True)
# Attempt to connect
conn.connect()
# Get a file-like object to access to the stream
stream = conn.create_stream()
# Read 1024 bytes of data
data = stream.read(1024)

This is the result when I print data:

b'FLV\x01\x05\x00\x00\x00\t\x00\x00\x00\x00\x12\x00\x01M\x00\x00\x00\x00\x00\x00\x00\x02\x00\nonMetaData\x03\x00\x08duration\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x08filesize\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05width\x00@\x94\x00\x00\x00\x00\x00\x00\x00\x06height\x00@\x86\x80\x00\x00\x00\x00\x00\x00\x0cvideocodecid\x02\x00\x04avc1\x00\rvideodatarate\x00@i\x00\x00\x00\x00\x00\x00\x00\tframerate\x00@>\x00\x00\x00\x00\x00\x00\x00\x0caudiocodecid\x02\x00\x04mp4a\x00\raudiodatarate\x00@`\x00\x00\x00\x00\x00\x00\x00\x0faudiosamplerate\x00@\xe7p\x00\x00\x00\x00\x00\x00\x0faudiosamplesize\x00@0\x00\x00\x00\x00\x00\x00\x00\raudiochannels\x00@\x00\x00\x00\x00\x00\x00\x00\x00\x06stereo\x00?\xf0\x00\x00\x00\x00\x00\x00\x00\x07encoder\x02\x00\x1eRogue Amoeba RTMP Streamer 1.0\x00\x00\t\x00\x00\x01X\x08\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\xaf\x00\x11\x90\x00\x00\x00\x0f'

I want to convert this to PCM or WAV format. How can I do this?

I don't know how to convert media with python, this code is just to parse some metadata. I think that you can save the data to a local file and then use some specialized tool for media conversion like ffmpeg.