Project

General

Profile

Statistics
| Branch: | Revision:

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

History | View | Annotate | Download (3.12 KB)

1
#!/usr/bin/env python
2

    
3
import gi
4
gi.require_version('Gst', '1.0')
5
gi.require_version('GstBase', '1.0')
6

    
7
from gi.repository import GObject, Gst, GstBase
8

    
9
import os
10
import gdp
11
import sys
12
import cPickle
13

    
14
GObject.threads_init()
15
Gst.init(None)
16

    
17
class GDPSrc(GstBase.BaseSrc):
18

    
19
    __gtype_name__ = "GDPSrc"
20
    __gstmetadata__ = ('Global Data Plane Source',
21
                        'GDPSrc',
22
                        'A source element to read data from a GDP log',
23
                        'Nitesh Mor')
24
    __gsttemplates__ = Gst.PadTemplate.new("src",
25
                                            Gst.PadDirection.SRC,
26
                                            Gst.PadPresence.ALWAYS,
27
                                            Gst.Caps.new_any())
28

    
29
    __gproperties__ = {
30
                "logname": (str, "GDP Logname",
31
                           "Name of GDP log that we should read from",
32
                           "logname_not_set",
33
                            GObject.PARAM_READWRITE),
34
                      }
35

    
36
    def __init__(self, *args):
37

    
38
        print("initializing")
39

    
40
        GstBase.BaseSrc.__init__(self, *args)
41
        self.set_format(Gst.Format.BUFFERS)
42
        self.set_blocksize(65000)
43

    
44
        self.logname = None
45
        self.lh = None
46

    
47
        self.cap = None
48
        self.buflist = []
49

    
50
    def do_set_property(self, prop, value):
51
        if prop.name == "logname":
52
            self.logname = value
53
        else:
54
            raise AttributeError
55

    
56
    def do_get_property(self, prop):
57
        if prop.name == "logname":
58
            return self.logname
59
        else:
60
            raise AttributeError
61

    
62
    def do_start(self):
63
        print "do_start"
64

    
65
        self.lh = gdp.GDP_GCL(gdp.GDP_NAME(self.logname), gdp.GDP_MODE_RO)
66
        self.lh.subscribe(1, 0, None)
67

    
68
        new_event = Gst.Event.new_stream_start("some-stream")
69
        self.send_event(new_event)
70
        return True
71

    
72

    
73
    def do_get_caps(self, caps):
74
        print "do_get_caps", caps, self.cap
75
        return self.cap
76

    
77

    
78
    def do_event(self, event):
79
        print "do_event", event, event.type
80
        return True
81

    
82

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

    
85
        data = self.__load_data()
86
        if data is None:
87
            return Gst.FlowReturn.EOS
88

    
89
        assert buf.get_size()>=len(data)
90
        buf.resize(0, len(data))
91
        ret = buf.fill(0, data)
92
        return Gst.FlowReturn.OK
93

    
94

    
95
    def __load_data(self):
96

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

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

    
101
            event = self.lh.get_next_event(timeout)
102
            if event is None:
103
                return None
104

    
105
            datum = event["datum"]
106
            (__cap, self.buflist) = cPickle.loads(datum["data"])
107
            self.cap = Gst.Caps.from_string(__cap)
108
            self.set_caps(self.cap)
109

    
110
        if len(self.buflist)>0:
111
            data = self.buflist[0]
112
            self.buflist = self.buflist[1:]
113
            return data
114
        else:
115
            return self.__load_data()
116

    
117

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

    
120
if __name__ == '__main__':
121

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