GStreamer Python Dynamic Switching

dynamic_switching.py

import gi
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst
from threading import Thread, Event

GObject.threads_init()
Gst.init(None)

class Main:
    def __init__(self):
        self.pipeline = Gst.Pipeline.new("pipeline")
        self.bus = self.pipeline.get_bus()

        self.videotestsrc1 = Gst.ElementFactory.make("videotestsrc", "videotestsrc1")
        self.videotestsrc1_src_pad = self.videotestsrc1.get_static_pad("src")
        self.videotestsrc1.set_property("pattern", 0)
        self.pipeline.add(self.videotestsrc1)

        self.videotestsrc2 = Gst.ElementFactory.make("videotestsrc", "videotestsrc2")
        self.videotestsrc2_src_pad = self.videotestsrc2.get_static_pad("src")
        self.videotestsrc2.set_property("pattern", 1)
        self.pipeline.add(self.videotestsrc2)

        self.input_selector = Gst.ElementFactory.make("input-selector", "input-selector")
        self.input_selector_sink_pad_1 = self.input_selector.get_request_pad("sink_1")
        self.input_selector_sink_pad_2 = self.input_selector.get_request_pad("sink_2")
        self.input_selector.set_property("active-pad", self.input_selector_sink_pad_1)
        self.pipeline.add(self.input_selector)

        self.autovideosink = Gst.ElementFactory.make("autovideosink", "autovideosink")
        self.pipeline.add(self.autovideosink)

        self.videotestsrc1_src_pad.link(self.input_selector_sink_pad_1)
        self.videotestsrc2_src_pad.link(self.input_selector_sink_pad_2)

        self.input_selector.link(self.autovideosink)

    def change_source(self):
        while not self.stop_event.is_set():
            self.stop_event.wait(1)

            if self.stop_event.is_set():
                break

            if self.input_selector_sink_pad_1.get_property("active"):
                self.input_selector.set_property("active-pad", self.input_selector_sink_pad_2)
            else:
                self.input_selector.set_property("active-pad", self.input_selector_sink_pad_1)

    def run(self):
        self.pipeline.set_state(Gst.State.PLAYING)

        self.stop_event = Event()
        self.change_source_thread = Thread(target=self.change_source)
        self.change_source_thread.start()

        while True:
            try:
                message = self.bus.timed_pop(Gst.SECOND)

                if message == None:
                    pass
                elif message.type == Gst.MessageType.EOS:
                    break
                elif message.type == Gst.MessageType.ERROR:
                    break
            except KeyboardInterrupt:
                break

        self.stop_event.set()
        self.pipeline.set_state(Gst.State.NULL)

start = Main()
start.run()

Thanks for sharing, but is there any hint how to use this snippet practically?

Running it "as is" with Python3 gives

/Users/decades/Documents/tmp/appsrctest/test.py:6: PyGIDeprecationWarning: Since version 3.11, calling threads_init is no longer needed. See: https://wiki.gnome.org/PyGObject/Threading
  GObject.threads_init()
Traceback (most recent call last):
  File "/Users/decades/Documents/tmp/appsrctest/test.py", line 73, in <module>
    start = Main()
  File "/Users/decades/Documents/tmp/appsrctest/test.py", line 15, in __init__
    self.videotestsrc1_src_pad = self.videotestsrc1.get_static_pad("src")
AttributeError: 'NoneType' object has no attribute 'get_static_pad'

self.videotestsrc1 = Gst.ElementFactory.make("videotestsrc", "videotestsrc1")

self.videotestsrc1 is None after this...

OK, disregard please. There is a general problem with the GST Python bindings on MacOS 12.4 M1