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/python/psd.py

@@ -55,6 +55,7 @@ from gstlal import datasource
 from gstlal import pipeparts
 from gstlal import pipeio
 from gstlal import simplehandler
+from gstlal.stream import MessageType, Stream
 
 
 __doc__ = """
@@ -95,6 +96,15 @@ class PSDHandler(simplehandler.Handler):
 		return False
 
 
+class PSDTracker:
+	def __init__(self):
+		self.psd = None
+
+	def on_spectrum_message(self, message):
+		self.psd = pipeio.parse_spectrum_message(message)
+		return True
+
+
 #
 # measure_psd()
 #
@@ -138,53 +148,46 @@ def measure_psd(gw_data_source_info, instrument, rate, psd_fft_length = 8, verbo
 		raise ValueError("segment %s too short" % str(gw_data_source_info.seg))
 
 	#
-	# build pipeline
+	# calculate number of samples to average over
 	#
 
-	if verbose:
-		print("measuring PSD in segment %s" % str(gw_data_source_info.seg), file=sys.stderr)
-		print("building pipeline ...", file=sys.stderr)
-	mainloop = GObject.MainLoop()
-	pipeline = Gst.Pipeline(name="psd")
-	handler = PSDHandler(mainloop, pipeline)
-
-	head, _, _ = datasource.mkbasicsrc(pipeline, gw_data_source_info, instrument, verbose = verbose)
-	head = pipeparts.mkcapsfilter(pipeline, head, "audio/x-raw, rate=[%d,MAX]" % rate)	# disallow upsampling
-	head = pipeparts.mkresample(pipeline, head, quality = 9)
-	head = pipeparts.mkcapsfilter(pipeline, head, "audio/x-raw, rate=%d" % rate)
-	head = pipeparts.mkqueue(pipeline, head, max_size_buffers = 8)
 	if gw_data_source_info.seg is not None:
 		average_samples = int(round(float(abs(gw_data_source_info.seg)) / (psd_fft_length / 2.) - 1.))
 	else:
 		#FIXME maybe let the user specify this
 		average_samples = 64
-	head = pipeparts.mkwhiten(pipeline, head, psd_mode = 0, zero_pad = 0, fft_length = psd_fft_length, average_samples = average_samples, median_samples = 7)
-	pipeparts.mkfakesink(pipeline, head)
 
 	#
-	# setup signal handler to shutdown pipeline for live data
+	# initialize PSD tracker
 	#
 
-	if gw_data_source_info.data_source in ("lvshm", "framexmit"):# FIXME what about nds online?
-		simplehandler.OneTimeSignalHandler(pipeline)
+	tracker = PSDTracker()
 
 	#
-	# process segment
+	# build pipeline
 	#
 
 	if verbose:
-		print("putting pipeline into READY state ...", file=sys.stderr)
-	if pipeline.set_state(Gst.State.READY) == Gst.StateChangeReturn.FAILURE:
-		raise RuntimeError("pipeline failed to enter READY state")
-	if gw_data_source_info.data_source not in ("lvshm", "framexmit"):# FIXME what about nds online?
-		datasource.pipeline_seek_for_gps(pipeline, *gw_data_source_info.seg)
-	if verbose:
-		print("putting pipeline into PLAYING state ...", file=sys.stderr)
-	if pipeline.set_state(Gst.State.PLAYING) == Gst.StateChangeReturn.FAILURE:
-		raise RuntimeError("pipeline failed to enter PLAYING state")
+		print("measuring PSD in segment %s" % str(gw_data_source_info.seg), file=sys.stderr)
+		print("building pipeline ...", file=sys.stderr)
+
+	stream = Stream.from_datasource(gw_data_source_info, instrument, verbose=verbose)
+	stream.add_callback(MessageType.ELEMENT, "spectrum", tracker.on_spectrum_message)
+
+	stream = stream.capsfilter(f"audio/x-raw, rate=[{rate:d},MAX]")  # disallow upsampling
+	stream.resample(quality=9) \
+		.capsfilter(f"audio/x-raw, rate={rate:d}") \
+		.queue(max_size_buffers=8) \
+		.whiten(psd_mode=0, zero_pad=0, fft_length=psd_fft_length, average_samples=average_samples, median_samples=7) \
+		.fakesink()
+
+	#
+	# process segment
+	#
+
 	if verbose:
 		print("running pipeline ...", file=sys.stderr)
-	mainloop.run()
+	stream.start()
 
 	#
 	# done
@@ -192,7 +195,7 @@ def measure_psd(gw_data_source_info, instrument, rate, psd_fft_length = 8, verbo
 
 	if verbose:
 		print("PSD measurement complete", file=sys.stderr)
-	return handler.psd
+	return tracker.psd
 
 
 def read_psd(filename: str, verbose: Optional[bool] = False) -> Dict[str, lal.REAL8FrequencySeries]: