Plot horizon distance from ranking statistics

This MR will change the horizon distance plotting program in the DAG: instead of creating one horizon distance plot from reference PSDs, the new program create the horizon distance from the ranking statistic for each template bank bin. The main changes that affect the DAG workflow are summarized below:

• gstlal_inspiral_pipe: change plotting program from gstlal_plot_psd_horizon to gstlal_inspiral_plot_rankingstats_horizon (previously called gstlal_inspiral_plot_psd_horizon which does not accurately describe what it actually does, also see miscellaneous changes below).
• inspiral.py: update the horizon_dist_layer for the new plotting program.
• gstlal_inspiral_summary_page: horizon distances plots are put inside a new UI component (report.ImagePagination) so that all the plots will not be fetched at the same time and can be paginated.
• make the above changes to new style DAG workflow as well.

Miscellaneous changes:

• add new plotting module called horizon.py
• update the gstlal_plot_psd_horizon to use horizon.py module,
• renamed gstlal_inspiral_plot_psd_horizon --> gstlal_inspiral_plot_rankingstats_horizon

Note: this MR depends on gstlal-visualisation/ligo-scald!36 (merged), it's better not to merged before that request has been merged.

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()