/usr/lib/plainbox-providers-1/checkbox/bin/gst_pipeline_test is in plainbox-provider-checkbox 0.3-2.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
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 70 71 72 73 74 75 76 77 78 | #!/usr/bin/env python3
from argparse import ArgumentParser
import logging
import re
import os
import sys
import time
from gi.repository import Gst
from gi.repository import GLib
from subprocess import check_output
def check_state(device):
"""Checks whether the sink is available for the given device.
"""
sink_info = check_output(['pacmd', 'list-sinks'],
universal_newlines=True)
data = sink_info.split("\n")
try:
device_name = re.findall(".*name:\s.*%s.*" % device, sink_info)[0].lstrip()
sink = re.findall(".*name:\s<(.*%s.*)>" % device, sink_info)[0].lstrip()
status = data[data.index("\t" + device_name) + 3]
except (IndexError, ValueError):
logging.error("Failed to find status for device: %s" % device)
return False
os.environ['PULSE_SINK'] = sink
logging.info("[ Pulse sink ]".center(80, '='))
logging.info("Device: %s %s" % (device_name.strip(), status.strip()))
return status
def main():
parser = ArgumentParser(description='Simple GStreamer pipeline player')
parser.add_argument('PIPELINE',
help='Quoted GStreamer pipeline to launch')
parser.add_argument('-t', '--timeout',
type=int, required=True,
help='Timeout for running the pipeline')
parser.add_argument('-d', '--device',
type=str,
help="Device to check for status")
args = parser.parse_args()
logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.INFO,
stream=sys.stdout)
exit_code = 0
if args.device:
if not check_state(args.device):
exit_code = 1
Gst.init(None)
try:
print("Attempting to initialize Gstreamer pipeline: {}".format(
args.PIPELINE))
element = Gst.parse_launch(args.PIPELINE)
except GLib.GError as error:
print("Specified pipeline couldn't be processed.")
print("Error when processing pipeline: {}".format(error))
#Exit harmlessly
return(2)
print("Pipeline initialized, now starting playback.")
element.set_state(Gst.State.PLAYING)
if args.timeout:
time.sleep(args.timeout)
element.set_state(Gst.State.NULL)
return exit_code
if __name__ == "__main__":
sys.exit(main())
|