gst-raiplay-download.sh: fix some shellcheck warnings
[experiments/gstreamer.git] / python / gst-decoupled-pipelines.py
1 #!/usr/bin/env python3
2
3 import sys
4
5 from math import pi
6
7 import gi
8 gi.require_version('Gst', '1.0')
9 from gi.repository import Gst
10 Gst.init(None)
11
12 from gi.repository import GObject
13 GObject.threads_init()
14
15
16 PIPELINE_SRC = """
17 videotestsrc pattern=18 background-color=4294901760 ! intervideosink
18 """
19
20 #PIPELINE_SRC = """
21 #v4l2src ! video/x-raw,width=320,height=240 ! intervideosink
22 #"""
23
24 PIPELINE_SINK = """
25 intervideosrc timeout=-1 ! videoconvert ! rotate name=rotate ! videoconvert ! autovideosink
26 """
27
28
29 class Player:
30     def __init__(self):
31         self.loop = GObject.MainLoop()
32         self.pipeline_src = Gst.parse_launch(PIPELINE_SRC)
33         self.pipeline_sink = Gst.parse_launch(PIPELINE_SINK)
34
35         self.rotate = self.pipeline_sink.get_by_name('rotate')
36         self.paused = 0
37         self.angle = 0
38
39         bus = self.pipeline_src.get_bus()
40         bus.add_signal_watch()
41         bus.connect('message::eos', self.on_eos)
42         bus.connect('message::error', self.on_error)
43         bus.connect('message::state-changed', self .on_state_changed)
44
45     def run(self):
46         self.pipeline_src.set_state(Gst.State.PLAYING)
47         self.pipeline_sink.set_state(Gst.State.PLAYING)
48         self.loop.run()
49
50     def quit(self):
51         self.pipeline_sink.set_state(Gst.State.NULL)
52         self.pipeline_src.set_state(Gst.State.NULL)
53         self.loop.quit()
54
55     def on_rotate(self):
56         self.angle -= pi / 100
57         self.angle %= 2 * pi
58         self.rotate.set_property("angle", self.angle)
59
60     def on_input(self):
61         print(self.paused)
62         if self.paused:
63             self.pipeline_src.set_state(Gst.State.PLAYING)
64         else:
65             self.pipeline_src.set_state(Gst.State.PAUSED)
66
67         self.paused ^= 1
68
69     def on_eos(self, bus, msg):
70         self.quit()
71
72     def on_error(self, bus, msg):
73         (err, debug) = msg.parse_error()
74         print("Error: %s" % err)
75         self.quit()
76
77     def on_state_changed(self, bus, msg):
78         if msg.src != self.pipeline_src:
79             return
80
81         old_state, new_state, pending = msg.parse_state_changed()
82         print("%s from %s to %s" % (msg.src.get_name(), old_state, new_state))
83
84
85 def main():
86     player = Player()
87
88     def stdin_cb(source, condition):
89         source.readline()
90         player.on_input()
91         return True
92
93     def timeout_cb():
94         player.on_rotate()
95         return True
96
97     GObject.io_add_watch(sys.stdin, GObject.IO_IN, stdin_cb)
98     GObject.timeout_add(100, timeout_cb)
99
100     print("\nPress Enter to freeze the video\n")
101     player.run()
102
103
104 if __name__ == '__main__':
105     sys.exit(main())