Project

General

Profile

Statistics
| Branch: | Revision:

gdp-if / gstreamer-log-plugin / gdpsink.py @ master

History | View | Annotate | Download (3.11 KB)

1
#!/usr/bin/env python
2

    
3
# Writes GStreamer capabilities and buffers to GDP logs as records
4

    
5
# Internally, each GDP record is a serialized tuple (cap, buflist), where
6
# buflist is a list of buffers, all sharing the same cap.
7

    
8

    
9
import gi
10
gi.require_version('Gst', '1.0')
11
gi.require_version('GstBase', '1.0')
12

    
13
from gi.repository import GObject, Gst, GstBase
14
import os
15
import cPickle
16
import gdp
17
import sys
18
import time
19

    
20
## Params to tune number of buffers in a single GDP record. A record
21
## should *typically* not exceed any of these limits.
22
MAXBUFS = 1000
23
MAXBYTES = 65000
24
MAXTIME = 1.0
25

    
26
GObject.threads_init()
27
Gst.init(None)
28

    
29
class GDPSink(GstBase.BaseSink):
30

    
31
    __gtype_name__ = "GDPSink"
32
    __gstmetadata__ = ('Global Data Plane Sink',
33
                       'GDPSink',
34
                       'A sink element to store data in a GDP log',
35
                       'Nitesh Mor')
36

    
37
    __gsttemplates__ = Gst.PadTemplate.new("sink",
38
                                           Gst.PadDirection.SINK,
39
                                           Gst.PadPresence.ALWAYS,
40
                                           Gst.Caps.new_any())
41

    
42
    __gproperties__ = {
43
                "logname": (str, "GDP Logname",
44
                           "Name of GDP log where data should go to",
45
                           "logname_not_set",
46
                            GObject.PARAM_READWRITE),
47
                      }
48

    
49
    def __init__(self, *args):
50

    
51
        GstBase.BaseSink.__init__(self, *args)
52

    
53
        self.buflist = []
54
        self.buflist_size = 0
55
        self.last_write = time.time()
56

    
57
        self.logname = None
58
        self.lh = None
59

    
60

    
61
    def do_set_property(self, prop, value):
62
        if prop.name == "logname":
63
            self.logname = value
64
        else:
65
            raise AttributeError
66

    
67
    def do_get_property(self, prop):
68
        if prop.name == "logname":
69
            return self.logname
70
        else:
71
            raise AttributeError
72

    
73

    
74
    def do_start(self):
75
        print "starting", self.logname
76
        self.lh = gdp.GDP_GCL(gdp.GDP_NAME(self.logname), gdp.GDP_MODE_AO)
77
        return True
78

    
79

    
80
    def do_event(self, event):
81
        print "do_event", event, event.type, event.seqnum, event.timestamp
82
        return True
83

    
84

    
85
    def do_render(self, buf):
86
        all_data = buf.extract_dup(0, buf.get_size())
87
        self.__dump_data(all_data)
88
        return Gst.FlowReturn.OK
89

    
90

    
91
    def __dump_data(self, data):
92

    
93
        cur_time = time.time()
94

    
95
        if (len(self.buflist)>=MAXBUFS or \
96
                self.buflist_size + len(data) >= MAXBYTES or \
97
                cur_time - self.last_write >= MAXTIME):
98

    
99
            ## We should write the current data to a record
100
            caps = self.sinkpad.get_current_caps().to_string()
101
            __data = cPickle.dumps((caps, self.buflist))
102

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

    
105
            self.last_write = cur_time
106
            self.buflist = []
107
            self.buflist_size = 0
108

    
109
        self.buflist.append(data)
110
        self.buflist_size += len(data)
111

    
112

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

    
115
if __name__ == '__main__':
116

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