gstlal-ugly/bin/gstlal_ifo_stat: use gstlal.datasource and only allow one ifo

gstlal-ugly/bin/gstlal_ifo_stat

@@ -30,6 +30,7 @@ Gst.init(None)
 from optparse import OptionParser
 from collections import deque, defaultdict
 
+from gstlal import datasource
 from gstlal import simplehandler
 from gstlal import pipeparts
 
@@ -39,9 +40,10 @@ from ligo.scald.io import influx
 def parse_command_line():
 	parser = OptionParser(description = __doc__)
 
+	# generic "source" options
+	datasource.append_options(parser)
+
 	parser.add_option("--output-url", metavar = "path", help = "Write to this url")
-	parser.add_option("--channel-name", metavar="channel", type = str, action = 'append', help = "Specify channel names for each ifo as --channel-name=ifo:channel-name. Can be given multiple times.")
-	parser.add_option("--shared-memory-directory", metavar="path", type = str, action = 'append', help = "Specify the shared memory directory to read frame files from as --shared-memory-directory=ifo:dir. Can be given multiple times.")
 	parser.add_option("--wait-time", metavar = int, default = 120, help = "wait time")
 	parser.add_option("--scald-config", metavar = "file", help = "sets ligo-scald options based on yaml configuration.")
 	parser.add_option("--verbose", action="store_true", help = "Be verbose.")
@@ -52,8 +54,9 @@ def parse_command_line():
 
 
 class IFOStatusTracker(object):
-	def __init__(self, options):
+	def __init__(self, options, ifo):
 		logging.info('setting up ifo status tracker...')
+		self.ifo = ifo
 		self.output_url = options.output_url
 		self.last_reduce = None
 
@@ -72,6 +75,7 @@ class IFOStatusTracker(object):
 		self.previous = defaultdict(lambda: (None, None))
 		self.duration = defaultdict(lambda: 0)
 
+
 	@staticmethod
 	def LIGO_parse_state(bitmask):
 		hoftbit = int(bitmask[31])
@@ -100,7 +104,7 @@ class IFOStatusTracker(object):
 
 		return state
 
-	def bufhandler(self, elem, ifo):
+	def bufhandler(self, elem):
 		buf = elem.emit("pull-sample").get_buffer()
 		(result, mapinfo) = buf.map(Gst.MapFlags.READ)
 		if mapinfo.data:
@@ -110,13 +114,13 @@ class IFOStatusTracker(object):
 			bit = int(bit)
 			time = float(time)
 
-			if ifo == "H1" or ifo == "L1":
+			if self.ifo == "H1" or self.ifo == "L1":
 				bitmask = bin(bit)
 				npad = 32 - len(bitmask)
 				bitmask = "0" * npad + bitmask
 				state = self.LIGO_parse_state(bitmask)
 
-			elif ifo == "K1":
+			elif self.ifo == "K1":
 				# only check 0th bit
 				state = "Observing" if (bit & 0b1 == 0b1) else "Down"
 			else:
@@ -132,19 +136,19 @@ class IFOStatusTracker(object):
 			statechange = False
 
 			## track duration
-			if state == self.previous[ifo][1]:
-				self.duration[ifo] += time - self.previous[ifo][0]
-			elif self.previous[ifo][0]:
+			if state == self.previous[self.ifo][1]:
+				self.duration[self.ifo] += time - self.previous[self.ifo][0]
+			elif self.previous[self.ifo][0]:
 				statechange = True
-				self.duration[ifo] = time - self.previous[ifo][0]
+				self.duration[self.ifo] = time - self.previous[self.ifo][0]
 
-			logging.debug(f'{ifo} h(t) state at {time} = {state} | duration = {self.duration[ifo]}')
+			logging.debug(f'{self.ifo} h(t) state at {time} = {state} | duration = {self.duration[self.ifo]}')
 
 			## store data
-			self.timedeq[ifo].append(int(time))
-			self.datadeq['state'][ifo].append(state)
-			self.datadeq['hoft_ok'][ifo].append(hoft_ok)
-			self.datadeq['duration'][ifo].append(self.duration[ifo])
+			self.timedeq[self.ifo].append(int(time))
+			self.datadeq['state'][self.ifo].append(state)
+			self.datadeq['hoft_ok'][self.ifo].append(hoft_ok)
+			self.datadeq['duration'][self.ifo].append(self.duration[self.ifo])
 
 			if self.last_reduce is None:
 				self.last_reduce = time
@@ -152,7 +156,7 @@ class IFOStatusTracker(object):
 			## write to influx
 			if time - self.last_reduce >= 50. or statechange:
 				if statechange:
-					logging.info(f'{ifo} h(t) state changed to {state} | time since last reduced = {time-self.last_reduce}')
+					logging.info(f'{self.ifo} h(t) state changed to {state} | time since last reduced = {time-self.last_reduce}')
 				self.last_reduce = int(utils.gps_now())
 				logging.debug(f'Reducing data to influx...')
 
@@ -167,7 +171,7 @@ class IFOStatusTracker(object):
 				for route in outdata:
 					self.influx_sink.store_columns(route, outdata[route], aggregate = "max")
 
-			self.previous[ifo] = (time, state)
+			self.previous[self.ifo] = (time, state)
 
 		buf.unmap(mapinfo)
 		return Gst.FlowReturn.OK
@@ -175,44 +179,33 @@ class IFOStatusTracker(object):
 def main():
 	options, filenames = parse_command_line()
 
-	channel_dict = {}
-	dirs_dict = {}
-	for channel in options.channel_name:
-		ifo, channel_name = channel.split('=')
-		channel_dict.update({
-			ifo: channel_name
-                })
-
-	for dir in options.shared_memory_directory:
-		ifo, shm_dir = dir.split('=')
-		dirs_dict.update({
-			ifo: shm_dir
-                })
-
-	if not channel_dict.keys() == dirs_dict.keys():
-		raise Exception(f'List of IFOs provided with channels does not match list of IFOs provided with SHM directories.')
-
 	# set up logging
 	log_level = logging.DEBUG if options.verbose else logging.INFO
 	logging.basicConfig(level = log_level, format = "%(asctime)s | gstlal_gw_stat : %(levelname)s : %(message)s")
 
-	tracker = IFOStatusTracker(options)
+	gw_data_source_info = datasource.DataSourceInfo.from_optparse(options)
+
+	# only support one channel
+	if len(gw_data_source_info.channel_dict.keys()) > 1:
+		logging.warning("Provided more than one instrument, but only one is supporting. Defaulting to use the first provided.")
+	ifo = list(gw_data_source_info.channel_dict.keys())[0]
+
+	tracker = IFOStatusTracker(options, ifo)
 
 	# building the event loop and pipeline
 	mainloop = GObject.MainLoop()
 	pipeline = Gst.Pipeline("gstlal_dq")
 	handler = simplehandler.Handler(mainloop, pipeline)
 
-	for ifo in channel_dict.keys():
-		src = pipeparts.mkdevshmsrc(pipeline, shm_dirname = dirs_dict[ifo], wait_time = options.wait_time)
-		src = pipeparts.mkframecppchanneldemux(pipeline, src, do_file_checksum = False, skip_bad_files = True)
-		pipeparts.framecpp_channeldemux_set_units(src, {f"{ifo}:{channel_dict[ifo]}": "strain"})
-		state = pipeparts.mkqueue(pipeline, None, max_size_buffers = 0, max_size_bytes = 0, max_size_time = Gst.SECOND * 60 * 1) # 1 minutes of buffering
-		pipeparts.src_deferred_link(src, f"{ifo}:{channel_dict[ifo]}", state.get_static_pad("sink"))
-		sink = pipeparts.mkgeneric(pipeline, state, "lal_nxydump")
-		sink = pipeparts.mkprogressreport(pipeline, sink, name=ifo)
-		sink = pipeparts.mkappsink(pipeline, sink, max_buffers = 1)
-		sink.connect("new-sample", tracker.bufhandler, ifo)
+	src = pipeparts.mkdevshmsrc(pipeline, shm_dirname = gw_data_source_info.shared_memory_dir[ifo], wait_time = options.wait_time)
+	src = pipeparts.mkframecppchanneldemux(pipeline, src, do_file_checksum = False, skip_bad_files = True)
+	pipeparts.framecpp_channeldemux_set_units(src, {f"{ifo}:{gw_data_source_info.channel_name[ifo]}": "strain"})
+	state = pipeparts.mkqueue(pipeline, None, max_size_buffers = 0, max_size_bytes = 0, max_size_time = Gst.SECOND * 60 * 1) # 1 minutes of buffering
+	pipeparts.src_deferred_link(src, f"{ifo}:{gw_data_source_info.channel_name[ifo]}", state.get_static_pad("sink"))
+	sink = pipeparts.mkgeneric(pipeline, state, "lal_nxydump")
+	sink = pipeparts.mkprogressreport(pipeline, sink, name=ifo)
+	sink = pipeparts.mkappsink(pipeline, sink, max_buffers = 1)
+	sink.connect("new-sample", tracker.bufhandler)
 
 	if not pipeline.set_state(Gst.State.PLAYING):
 		raise RuntimeError("pipeline failed to enter PLAYING state")