GDP

#!/usr/bin/env python

# Writes GStreamer capabilities and buffers to GDP logs as records

# Internally, each GDP record is a serialized tuple (cap, buflist), where

# buflist is a list of buffers, all sharing the same cap.

import gi

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

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

from gi.repository import GObject, Gst, GstBase

import os

import cPickle

import gdp

import sys

import time

## Params to tune number of buffers in a single GDP record. A record

## should *typically* not exceed any of these limits.

MAXBUFS = 1000

MAXBYTES = 65000

MAXTIME = 1.0

GObject.threads_init()

Gst.init(None)

class GDPSink(GstBase.BaseSink):

    __gtype_name__ = "GDPSink"

    __gstmetadata__ = ('Global Data Plane Sink',

                       'GDPSink',

                       'A sink element to store data in a GDP log',

                       'Nitesh Mor')

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

                                           Gst.PadDirection.SINK,

                                           Gst.PadPresence.ALWAYS,

                                           Gst.Caps.new_any())

    __gproperties__ = {

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

                           "Name of GDP log where data should go to",

                           "logname_not_set",

                            GObject.PARAM_READWRITE),

                      }

    def __init__(self, *args):

        GstBase.BaseSink.__init__(self, *args)

        self.buflist = []

        self.buflist_size = 0

        self.last_write = time.time()

        self.logname = None

        self.lh = None

    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 "starting", self.logname

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

        return True

    def do_event(self, event):

        print "do_event", event, event.type, event.seqnum, event.timestamp

        return True

    def do_render(self, buf):

        all_data = buf.extract_dup(0, buf.get_size())

        self.__dump_data(all_data)

        return Gst.FlowReturn.OK

    def __dump_data(self, data):

        cur_time = time.time()

        if (len(self.buflist)>=MAXBUFS or \

                self.buflist_size + len(data) >= MAXBYTES or \

                cur_time - self.last_write >= MAXTIME):

            ## We should write the current data to a record

            caps = self.sinkpad.get_current_caps().to_string()

            __data = cPickle.dumps((caps, self.buflist))

            self.lh.append({"data":__data})

            self.last_write = cur_time

            self.buflist = []

            self.buflist_size = 0

        self.buflist.append(data)

        self.buflist_size += len(data)

Gst.Element.register(None, "gdpsink", Gst.Rank.NONE, GDPSink)

if __name__ == '__main__':

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

    sys.exit(-1)