gstlal_stage_frames: convert to stream API

gstlal/bin/gstlal_stage_frames

@@ -21,18 +21,12 @@ from optparse import OptionParser
 import os
 import sys
 
-import gi
-gi.require_version('Gst', '1.0')
-from gi.repository import GObject, Gst
-GObject.threads_init()
-Gst.init(None)
-
 import lal
 
 from gstlal import pipeparts
 from gstlal import simplehandler
 from gstlal import datasource
-from gstlal import multirate_datasource
+from gstlal.stream import Stream
 from ligo.lw import utils as ligolw_utils
 
 
@@ -56,9 +50,9 @@ def parse_command_line():
 	parser.add_option("--sample-rate", metavar = "Hz", default = 16384, type = "int", help = "Sample rate at which to generate the data, should be less than or equal to the sample rate of the measured psds provided, default = 16384 Hz, max 16384 Hz")
 	parser.add_option("--output-path", metavar = "name", default = ".", help = "Path to output frame files (default = \".\").")
 	parser.add_option("--output-channel-name", metavar = "name", help = "The name of the channel in the output frames. The default is the same as the channel name")
-	parser.add_option("--frame-type", metavar = "name", help = "Frame type, required")
-	parser.add_option("--frame-duration", metavar = "s", default = 16, type = "int", help = "Set the duration of the output frames.  The duration of the frame file will be multiplied by --frames-per-file.  Default: 16s")
-	parser.add_option("--frames-per-file", metavar = "n", default = 256, type = "int", help = "Set the number of frames per file.  Default: 256")
+	parser.add_option("--output-frame-type", metavar = "name", help = "Frame type, required")
+	parser.add_option("--output-frame-duration", metavar = "s", default = 16, type = "int", help = "Set the duration of the output frames.  The duration of the frame file will be multiplied by --frames-per-file.  Default: 16s")
+	parser.add_option("--output-frames-per-file", metavar = "n", default = 256, type = "int", help = "Set the number of frames per file.  Default: 256")
 	parser.add_option("-v", "--verbose", action = "store_true", help = "Be verbose (optional).")
 
 	#
@@ -70,8 +64,8 @@ def parse_command_line():
 	if options.sample_rate > 16384:
 		raise ValueError("--sample-rate must be <= 16384")
 
-	if options.frame_type is None:
-		raise ValueError("--frame-type is required")
+	if options.output_frame_type is None:
+		raise ValueError("--output-frame-type is required")
 
 
 	return options, filenames
@@ -83,76 +77,44 @@ def parse_command_line():
 
 options, filenames = parse_command_line()
 
+## Create output directory
+os.makedirs(options.output_path, exist_ok=True)
+
 ## Parse the command line options into a python.datasource.GWDataSourceInfo class instance
 gw_data_source = datasource.DataSourceInfo.from_optparse(options)
 
 ## Assume instrument is the first and only key of the python.datasource.GWDataSourceInfo.channel_dict
 instrument = list(gw_data_source.channel_dict.keys())[0]
 
-# disable progress reports if not verbose
-if not options.verbose:
-	pipeparts.mkprogressreport = lambda pipeline, src, *args: src
-
 # set default output channel if not set by user
 if options.output_channel_name is None:
 	options.output_channel_name = gw_data_source.channel_dict[instrument]
 
-## Setup the pipeline
-pipeline = Gst.Pipeline(name=os.path.split(sys.argv[0])[1])
-
-## Main loop 
-mainloop = GObject.MainLoop()
-
-## An instance of the python.simplehandler.Handler class
-handler = simplehandler.Handler(mainloop, pipeline)
-
-## Set the pipeline head to basic input from datasource.mkbasicsrc()
+## Initialize stream from data source
 # FIXME: fake source causes problems when making large buffers, so block_size needs to be overwritten
 gw_data_source.block_size = 8 * options.sample_rate
-head, _, _ = datasource.mkbasicsrc(pipeline, gw_data_source, instrument, verbose = options.verbose)
+stream = Stream.from_datasource(gw_data_source, instrument, verbose=options.verbose)
 
-## Set the pipeline head to be verbose with pipeparts.mkprogressreport()
-head = pipeparts.mkprogressreport(pipeline, head, "frames")
+## Add progress report to stream if verbose
+if options.verbose:
+	stream = stream.progressreport("frames")
 
 ## Downsample to requested rate
-head = pipeparts.mkinterpolator(pipeline, head)
-caps = "audio/x-raw, rate=%d" % options.sample_rate
-head = pipeparts.mkcapsfilter(pipeline, head, caps)
-
-if not os.path.isdir(options.output_path):
-	try:
-		os.makedirs(options.output_path)
-	except:
-		print("Unable to make directory ", options.output_path, file=sys.stderr)
-		pass
-else:
-	print("Target output directory already exists.")
+caps = f"audio/x-raw, rate={options.sample_rate}"
+stream = stream.interpolator().capsfilter(caps)
 
 ## create frames
-head = pipeparts.mkframecppchannelmux(pipeline, {"%s:%s" % (instrument, options.output_channel_name): head}, frame_duration = options.frame_duration, frames_per_file = options.frames_per_file)
+stream = stream.framecppchannelmux(
+	channels=f"{instrument}:{options.output_channel_name}",
+	frame_duration=options.output_frame_duration,
+	frames_per_file=options.output_frames_per_file
+)
 
 ## Write the frames to disk
-head = pipeparts.mkframecppfilesink(pipeline, head, frame_type = options.frame_type)
+framesink = stream.framecppfilesink(frame_type=options.output_frame_type)
 
 # Put O(100000 s) frames in each directory
-head.connect("notify::timestamp", pipeparts.framecpp_filesink_ldas_path_handler, (options.output_path, 5))
+framesink.connect("notify::timestamp", pipeparts.framecpp_filesink_ldas_path_handler, (options.output_path, 5))
 
 # Run it
-if pipeline.set_state(Gst.State.READY) == Gst.StateChangeReturn.FAILURE:
-	raise RuntimeError("pipeline failed to enter READY state")
-datasource.pipeline_seek_for_gps(pipeline, gw_data_source.seg[0], gw_data_source.seg[1])
-if pipeline.set_state(Gst.State.PLAYING) == Gst.StateChangeReturn.FAILURE:
-	raise RuntimeError("pipeline failed to enter PLAYING state")
-
-## Debugging output
-if "GST_DEBUG_DUMP_DOT_DIR" in os.environ:
-	pipeparts.write_dump_dot(pipeline, "%s_PLAYING" % pipeline.get_name(), verbose = True)
-
-	## Setup a signal handler to intercept SIGINT in order to write the pipeline graph at ctrl+C before cleanly shutting down
-	class SigHandler(simplehandler.OneTimeSignalHandler):
-		def do_on_call(self, signum, frame):
-			pipeparts.write_dump_dot(pipeline, "%s_SIGINT" % pipeline.get_name(), verbose = True)
-
-	sighandler = SigHandler(pipeline)
-
-mainloop.run()
+stream.start()