Ubuntu Wiki

   1 import gi
   2 import sys
   3 
   4 gi.require_version('Gst', '1.0')
   5 from gi.repository import Gst, GObject, GLib
   6 
   7 def on_message(bus, message):
   8     m_type = message.type
   9     if m_type == Gst.MessageType.EOS:
  10         print("End of Stream (EOS).")
  11         loop.quit()
  12     elif m_type == Gst.MessageType.ERROR:
  13         err, debug = message.parse_error()
  14         print(f"Error: {err.message}")
  15         print(f"Debugging info: {debug}")
  16         loop.quit()
  17     return True
  18 
  19 # Initialize GStreamer
  20 Gst.init(None)
  21 
  22 # Build the pipeline string
  23 # videotestsrc: generates a test video pattern
  24 # autovideosink: an automatic sink that displays video in a window
  25 pipeline_string = "videotestsrc ! autovideosink"
  26 
  27 # Create the pipeline from the string
  28 pipeline = Gst.parse_launch(pipeline_string)
  29 
  30 if not pipeline:
  31     print("Failed to create pipeline.")
  32     sys.exit(1)
  33 
  34 # Get the bus and set up the message handler
  35 bus = pipeline.get_bus()
  36 bus.add_signal_watch()
  37 bus.connect("message", on_message)
  38 
  39 # Create and run the GLib main loop
  40 loop = GLib.MainLoop()
  41 
  42 print("Starting video pipeline. Close the window or press Ctrl+C to stop.")
  43 pipeline.set_state(Gst.State.PLAYING)
  44 
  45 try:
  46     loop.run()
  47 except KeyboardInterrupt:
  48     print("Stopping pipeline.")
  49 finally:
  50     # Clean up
  51     pipeline.set_state(Gst.State.NULL)
  52     bus.remove_signal_watch()
  53     print("Pipeline stopped.")