GStreamer selector

Other / Gstreamer / gstreamer

dynamic selector
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
#!/usr/bin/env python3
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib

Gst.init(None)

# Define pipeline string using input-selector
pipeline_str = """
input-selector name=selector !
    videoconvert ! autovideosink
    v4l2src device=/dev/video0 ! videoconvert ! selector.sink_0
    v4l2src device=/dev/video4 ! videoconvert ! selector.sink_1
"""

print("Pipeline:\n", pipeline_str)
pipeline = Gst.parse_launch(pipeline_str)

selector = pipeline.get_by_name("selector")
pads = selector.sinkpads

# state for switching
current_index = {"i": 0}

def switch_source():
    current_index["i"] = (current_index["i"] + 1) % len(pads)
    pad = pads[current_index["i"]]
    selector.set_property("active-pad", pad)
    print(f"Switched to camera {current_index['i']}")
    return True  # keep repeating

# Set up bus to catch errors and EOS
bus = pipeline.get_bus()
bus.add_signal_watch()

def on_message(bus, message):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("Error:", err, debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        print("End of stream")
        loop.quit()

bus.connect("message", on_message)

# Start pipeline
pipeline.set_state(Gst.State.PLAYING)

# Switch every 5 seconds
GLib.timeout_add_seconds(5, switch_source)

loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    pipeline.set_state(Gst.State.NULL)
dynamic selector and compositor
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
#!/usr/bin/env python3
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib

Gst.init(None)

pipeline_str = """
    compositor name=final_comp sink_0::xpos=0 sink_0::ypos=0 sink_1::xpos=640 sink_1::ypos=0 !
        videoconvert ! autovideosink

    v4l2src device=/dev/video0 ! videoconvert ! tee name=tee0
    tee0. ! queue ! selector.sink_0
    tee0. ! queue ! videoscale ! video/x-raw,width=320,height=240 ! both_comp.sink_0

    v4l2src device=/dev/video4 ! videoconvert ! tee name=tee1
    tee1. ! queue ! selector.sink_1
    tee1. ! queue ! videoscale ! video/x-raw,width=320,height=240 ! both_comp.sink_1

    input-selector name=selector ! videoscale ! video/x-raw,width=640,height=480 ! final_comp.sink_0

    compositor name=both_comp sink_0::xpos=0 sink_0::ypos=0 sink_1::xpos=320 sink_1::ypos=0 !
        videoscale ! video/x-raw,width=640,height=480 ! final_comp.sink_1
"""



print("Pipeline:\n", pipeline_str)
pipeline = Gst.parse_launch(pipeline_str)

selector = pipeline.get_by_name("selector")
pads = selector.sinkpads

# Track which pad is active
current_index = {"i": 0}

def switch_source():
    current_index["i"] = (current_index["i"] + 1) % len(pads)
    pad = pads[current_index["i"]]
    selector.set_property("active-pad", pad)
    print(f"Switched selector to camera {current_index['i']}")
    return True  # repeat

bus = pipeline.get_bus()
bus.add_signal_watch()

def on_message(bus, message):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("Error:", err, debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        print("End of stream")
        loop.quit()

bus.connect("message", on_message)

pipeline.set_state(Gst.State.PLAYING)

# Switch every 5s
GLib.timeout_add_seconds(5, switch_source)

loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    pipeline.set_state(Gst.State.NULL)