i am trying to implement addition and deletion of sources at runtime for handling issue related to merging video sources from rtsp file stream , whole service stops running if one stream has error or stopped. can you provide python code for same addition and deletion of sources??
Also i tried using python code attached below. showing errors an empty screen running which hangs the system temporarily. please provide a relevant solution for the same.
Please find the code below.
import gi
gi.require_version(‘Gst’, ‘1.0’)
from gi.repository import GObject, Gst
from threading import Thread, Event
Gst.init(None)
class Main:
def init(self):
self.pipeline = Gst.Pipeline.new(“pipeline”)
self.bus = self.pipeline.get_bus()
self.current_pattern_type = 0
self.videotestsrc = Gst.ElementFactory.make("videotestsrc", "videotestsrc")
self.videotestsrc.set_property("pattern", self.current_pattern_type)
self.videotestsrc_src_pad = self.videotestsrc.get_static_pad("src")
self.pipeline.add(self.videotestsrc)
self.decodebin = Gst.ElementFactory.make("decodebin", "decodebin")
self.pipeline.add(self.decodebin)
self.decodebin.connect("pad-added", self.decodebin_src_pad_created)
self.autovideosink = Gst.ElementFactory.make("autovideosink", "autovideosink")
self.pipeline.add(self.autovideosink)
self.videotestsrc.link(self.decodebin)
def decodebin_src_pad_created(self, element, pad):
self.decodebin.link(self.autovideosink)
def custom_message(self, name):
custom_structure = Gst.Structure.new_empty(name)
custom_message = Gst.Message.new_application(None, custom_structure)
self.bus.post(custom_message)
def change_source(self):
while not self.stop_event.is_set():
self.stop_event.wait(1)
if self.stop_event.is_set():
break
self.custom_message("start_switch")
def videotestsrc_probe_callback(self, pad, info):
if self.start_switch:
if not self.videotestsrc_do_switch_sent:
self.videotestsrc_drop = True
self.custom_message("do_switch")
return Gst.PadProbeReturn.DROP
else:
self.videotestsrc_do_switch_sent = False
self.videotestsrc_drop = False
pad.remove_probe(info.id)
return Gst.PadProbeReturn.OK
def decodebin_probe_callback(self, pad, info):
if self.start_switch:
if not self.decodebin_do_switch_sent:
self.decodebin_drop = True
self.custom_message("do_switch")
return Gst.PadProbeReturn.DROP
else:
self.decodebin_do_switch_sent = False
self.decodebin_drop = False
pad.remove_probe(info.id)
return Gst.PadProbeReturn.OK
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()
self.start_switch = False
self.videotestsrc_drop = False
self.decodebin_drop = False
while True:
try:
message = self.bus.timed_pop(Gst.SECOND)
if message == None:
pass
elif message.type == Gst.MessageType.APPLICATION:
if message.get_structure().get_name() == "start_switch":
self.start_switch = True
self.videotestsrc_do_switch_sent = False
self.decodebin_do_switch_sent = False
self.videotestsrc.get_static_pad("src").add_probe(Gst.PadProbeType.BLOCK, self.videotestsrc_probe_callback)
self.decodebin.get_static_pad("src_0").add_probe(Gst.PadProbeType.BLOCK, self.decodebin_probe_callback)
elif message.get_structure().get_name() == "do_switch":
if self.videotestsrc_drop and self.decodebin_drop:
# Remove old videotestsrc.
self.videotestsrc.set_state(Gst.State.NULL)
self.videotestsrc.unlink(self.decodebin)
self.pipeline.remove(self.videotestsrc)
del self.videotestsrc
# Create new videotestsrc.
self.videotestsrc = Gst.ElementFactory.make("videotestsrc", "videotestsrc")
self.current_pattern_type = int(not(self.current_pattern_type))
self.videotestsrc.set_property("pattern", self.current_pattern_type)
self.pipeline.add(self.videotestsrc)
self.videotestsrc.link(self.decodebin)
self.videotestsrc.set_state(Gst.State.PLAYING)
self.start_switch = False
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()