GDP

#!/usr/bin/env python

import gi

gi.require_version('Gst', '1.0')

gi.require_version('GstBase', '1.0')

from gi.repository import GObject, Gst, GstBase

import os

import gdp

import sys

import cPickle

GObject.threads_init()

Gst.init(None)

class GDPSrc(GstBase.BaseSrc):

    __gtype_name__ = "GDPSrc"

    __gstmetadata__ = ('Global Data Plane Source',

                        'GDPSrc',

                        'A source element to read data from a GDP log',

                        'Nitesh Mor')

    __gsttemplates__ = Gst.PadTemplate.new("src",

                                            Gst.PadDirection.SRC,

                                            Gst.PadPresence.ALWAYS,

                                            Gst.Caps.new_any())

    __gproperties__ = {

                "logname": (str, "GDP Logname",

                           "Name of GDP log that we should read from",

                           "logname_not_set",

                            GObject.PARAM_READWRITE),

                      }

    def __init__(self, *args):

        print("initializing")

        GstBase.BaseSrc.__init__(self, *args)

        self.set_format(Gst.Format.BUFFERS)

        self.set_blocksize(65000)

        self.logname = None

        self.lh = None

        self.cap = None

        self.buflist = []

    def do_set_property(self, prop, value):

        if prop.name == "logname":

            self.logname = value

        else:

            raise AttributeError

    def do_get_property(self, prop):

        if prop.name == "logname":

            return self.logname

        else:

            raise AttributeError

    def do_start(self):

        print "do_start"

        self.lh = gdp.GDP_GCL(gdp.GDP_NAME(self.logname), gdp.GDP_MODE_RO)

        self.lh.subscribe(1, 0, None)

        new_event = Gst.Event.new_stream_start("some-stream")

        self.send_event(new_event)

        return True

    def do_get_caps(self, caps):

        print "do_get_caps", caps, self.cap

        return self.cap

    def do_event(self, event):

        print "do_event", event, event.type

        return True

    def do_fill(self, offset, size, buf):

        data = self.__load_data()

        if data is None:

            return Gst.FlowReturn.EOS

        assert buf.get_size()>=len(data)

        buf.resize(0, len(data))

        ret = buf.fill(0, data)

        return Gst.FlowReturn.OK

    def __load_data(self):

        timeout = {'tv_sec':1, 'tv_nsec':0, 'tv_accuracy':0.0}

        if self.cap is None or len(self.buflist)==0:

            event = self.lh.get_next_event(timeout)

            if event is None:

                return None

            datum = event["datum"]

            (__cap, self.buflist) = cPickle.loads(datum["data"])

            self.cap = Gst.Caps.from_string(__cap)

            self.set_caps(self.cap)

        if len(self.buflist)>0:

            data = self.buflist[0]

            self.buflist = self.buflist[1:]

            return data

        else:

            return self.__load_data()

Gst.Element.register(None, "gdpsrc", Gst.Rank.NONE, GDPSrc)

if __name__ == '__main__':

    sys.stderr.write("Use gst-launch-wrapper, do not use directly!")

    sys.exit(-1)