gst_loop_playback.py

"""
Loops video
Usage:
    python gst_loop_playback.py -l "filesrc location=video0.mp4 ! decodebin ! videoconvert ! gtksink sync=False" -r 2
"""


import os
import traceback
import argparse

import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstBase', '1.0')
from gi.repository import Gst, GObject, GstBase

Gst.init(None)

ap = argparse.ArgumentParser()
ap.add_argument("-l", "--gst_launch", required=True, help="Gst-Launch String")
ap.add_argument("-r", "--repeat", required=False, default=1, help="Max num of pipeline repetitions (loops)")

args = vars(ap.parse_args())


def create_dir(directory):
    if not os.path.exists(directory):
        os.makedirs(directory)


class UserData:

    def __init__(self, pipeline: Gst.Pipeline,
                 loop: GObject.MainLoop,
                 prerolled: bool = False,
                 max_loops_num: int = 1):

        self.pipeline = pipeline
        self.prerolled = prerolled
        self.current_loop_id = 1
        self.max_loops_num = max_loops_num

        self._loop = loop

    def quit(self):
        self._loop.quit()


def bus_message_handler(bus: Gst.Bus, message: Gst.Message, user_data: UserData):

    # global pipeline, prerolled, current_index, max_index

    mtype = message.type
    if mtype == Gst.MessageType.ASYNC_DONE:
        if not user_data.prerolled:
            user_data.pipeline.seek_simple(
                Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.SEGMENT, 0)
            user_data.prerolled = True
    elif mtype == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("{0}: {1}".format(err, debug))
        loop.quit()

    elif mtype == Gst.MessageType.WARNING:
        err, debug = message.parse_warning()
        print("{0}: {1}".format(err, debug))
        user_data.quit()

    elif mtype == Gst.MessageType.SEGMENT_DONE:
        if user_data.current_loop_id < user_data.max_loops_num:
            pipeline.seek_simple(Gst.Format.TIME, Gst.SeekFlags.SEGMENT, 0)
            user_data.current_loop_id += 1
            print(f"Loop... {user_data.current_loop_id}/{user_data.max_loops_num}")

        else:
            print("Video ended")
            user_data.quit()
    else:
        pass

    return True


pipeline = Gst.parse_launch(args["gst_launch"])

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

loop = GObject.MainLoop()

bus.connect("message", bus_message_handler, UserData(pipeline, loop, max_loops_num=int(args["repeat"])))
pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except:
    traceback.print_exc()

loop.quit()
pipeline.set_state(Gst.State.NULL)