import os
HASH_TEMPLATE = "%d_%d"
LOGNAME_TEMPLATE = "%s-%s"
START_DUR2PATH_TEMPLATE = "%s-%d-%d.%s"
GLOB2PATH_TEMPLATE = "%s-*-*.%s"
START2DIRMOD1E5_TEMPLATE = "%s-"
START_END2DIR_TEMPLATE = "%d_%d"
DQR_LOGNAME_TEMPLATE = "dqr-%s"
TAG2GROUP_TEMPLATE = "%s_%s"
# naming conventions for channel data products
PGLITCH = "PGLITCH"
LOGLIKE = "LOGLIKE"
EFF = "EFF"
FAP = "FAP"
RANK = "RANK"
OK = "OK"
def start_end2hash(start, end):
return HASH_TEMPLATE % (start, end)
def hash2start_end(string):
start, end = string.split("_")[:2]
return int(start), int(end)
def times_id2hash(start, end, data_id=None):
if data_id is None:
return f"{int(start)}_{int(end)}"
else:
return f"{int(start)}_{int(end)}_{data_id}"
def hash2id(hash_):
components = hash_.split("_")
if len(components) == 2:
return None
else:
return components[2]
[docs]def path2start_dur(path, suffix="trg"):
"""
extract the start and duration from a standard LIGO file name
>>> start, dur = path2start_dur('/path/to/H1-SNAX-1234560000-20.h5', suffix='h5')
>>> (start, dur)
(1234560000, 20)
"""
return (int(_) for _ in path[: -len(suffix) - 1].split("-")[-2:])
[docs]def path2start_end(path, suffix="trg"):
"""
extract the start and duration from a standard LIGO file name
>>> path2start_end('/path/to/H1-SNAX-1234560000-20.h5', suffix='h5')
(1234560000, 1234560020)
"""
start, dur = path2start_dur(path, suffix=suffix)
return start, start + dur
def start_dur2path(prefix, start, dur, suffix="trg"):
return START_DUR2PATH_TEMPLATE % (prefix, start, dur, suffix)
def start_end2path(prefix, start, end, suffix="trg"):
return start_dur2path(prefix, start, int(end) - int(start), suffix=suffix)
def glob2path(prefix, suffix="trg"):
return GLOB2PATH_TEMPLATE % (prefix, suffix)
[docs]def dirMOD1e52start_dur(directory):
"""
extract start and duration from directory name like
~/basename-${gpsMOD1e5}/
"""
return (
int(
os.path.basename(os.path.dirname(os.path.join(directory, ""))).split("-")[
-1
]
)
* 100000,
100000,
) # FIXME: FRAGILE
[docs]def dirMOD1e52start_end(directory):
"""
extract start and end from directory name like
~/basename-${gpsMOD1e5}/
"""
start, dur = dirMOD1e52start_dur(directory)
return start, start + dur
def start2dirMOD1e5_template(basename="basename", rootdir=""):
return os.path.join(rootdir, START2DIRMOD1E5_TEMPLATE % basename + "%d")
def start2dirMOD1e5(start, basename="basename", rootdir=""):
return start2dirMOD1e5_template(basename=basename, rootdir=rootdir) % (
int(start) / 100000
)
[docs]def dir2start_dur(directory):
"""
extract start and duration from directory like
~/${start}_${end}/
"""
start, end = dir2start_end(directory)
return start, end - start
[docs]def dir2start_end(directory):
"""
extract start and end from directory like
~/${start}_${end}/
"""
return tuple(
int(_)
for _ in os.path.basename(os.path.dirname(os.path.join(directory, ""))).split(
"_"
)
) # FIXME: FRAGILE
def start_end2dirtemplate(rootdir=""):
return os.path.join(rootdir, START_END2DIR_TEMPLATE)
def start_end2dir(start, end, rootdir=""):
return start_end2dirtemplate(rootdir=rootdir) % (start, end)
def start_dur2dir(start, dur, rootdir=""):
return start_end2dir(start, start + dur, rootdir=rootdir)
def start_end2fragmented_dir(start, end, rootdir=".", basename="START"):
return start_end2dir(
start, end, rootdir=start2dirMOD1e5(start, basename=basename, rootdir=rootdir)
)
def start_dur2fragmented_dir(start, dur, rootdir=".", basename="START"):
return start_end2fragmented_dir(
start, start + dur, rootdir=rootdir, basename=basename
)
[docs]def basename2snax_filename(basename):
"""
given a basename, will return a glob-compatible filename
compatible with a standard hdf5 set of features written to disk
"""
return basename + "-*.h5"
def channel2omicron_filename(instrument, channel, suffix="h5"):
return instrument + "-" + channel + "-*-*." + suffix
def tag2subdir(tag, task, rootdir=""):
return os.path.join(rootdir, task + "-" + tag)
def tag2traindir(tag, rootdir=""):
return tag2subdir(tag, "train", rootdir=rootdir)
def tag2evaluatedir(tag, rootdir=""):
return tag2subdir(tag, "evaluate", rootdir=rootdir)
def tag2calibratedir(tag, rootdir=""):
return tag2subdir(tag, "calibrate", rootdir=rootdir)
def tag2timeseriesdir(tag, rootdir=""):
return tag2subdir(tag, "timeseries", rootdir=rootdir)
def tag2reportdir(tag, rootdir=""):
return tag2subdir(tag, "report", rootdir=rootdir)
def tag2monitordir(tag, rootdir=""):
return tag2subdir(tag, "monitor", rootdir=rootdir)
def tag2batchdir(tag, rootdir=""):
return tag2subdir(tag, "batch", rootdir=rootdir)
def tag2streamdir(tag, rootdir=""):
return tag2subdir(tag, "stream", rootdir=rootdir)
def nickname2condorname(nickname):
return "condor" + nickname
def id2batchname(data_id):
return f"batch-{data_id}"
def id2nickname(nickname, data_id=None):
if data_id is not None:
return f"{nickname}_{data_id}"
else:
return nickname
def tag2group(tag, job):
return TAG2GROUP_TEMPLATE % (tag, job)
def fig2file(nickname, plot, start, dur, figtype="png"):
return start_dur2path("_".join([nickname, plot]), start, dur, suffix=figtype)
def channel_name_template(ifo, nickname, freq_min, freq_max):
return ifo + ":IDQ-%s_" + nickname.upper() + "_%d_%d" % (freq_min, freq_max)