diff --git a/gstlal-ugly/bin/gstlal_ifo_stat b/gstlal-ugly/bin/gstlal_ifo_stat
index 438cba9867761b6afa02cca70b36b28b108ac241..1cf1c20f49c9daab095f29853485530e13e43c1c 100755
--- a/gstlal-ugly/bin/gstlal_ifo_stat
+++ b/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")