# Copyright (C) 2020 Patrick Godwin (patrick.godwin@ligo.org)
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import json
import os
import sys
from ligo.segments import segment
from gstlal.config import Config as BaseConfig
from gstlal.config import dotdict, replace_keys
[docs]class Config(BaseConfig):
"""
Hold configuration used for inspiral-specific analyzes.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
# section-specific options
self.svd = dotdict(replace_keys(kwargs["svd"]))
self.filter = dotdict(replace_keys(kwargs["filter"]))
self.prior = dotdict(replace_keys(kwargs["prior"]))
self.rank = dotdict(replace_keys(kwargs["rank"]))
if "dcc" in kwargs:
self.dcc = dotdict(replace_keys(kwargs["dcc"]))
if "upload" in kwargs:
self.upload = dotdict(replace_keys(kwargs["upload"]))
if "pastro" in kwargs:
self.pastro = dotdict(replace_keys(kwargs["pastro"]))
if "metrics" in kwargs:
self.metrics = dotdict(replace_keys(kwargs["metrics"]))
if "services" in kwargs:
self.services = dotdict(replace_keys(kwargs["services"]))
if "summary" in kwargs:
self.summary = dotdict(replace_keys(kwargs["summary"]))
# set up analysis directories
if not self.data.analysis_dir:
self.data.analysis_dir = os.getcwd()
if not self.data.rerank_dir:
self.data.rerank_dir = self.data.analysis_dir
# validate config
self.validate_inspiral()
[docs] def validate_inspiral(self):
"""
Validate configuration settings.
"""
if self.svd.sub_banks:
for bank_name, params in self.svd.sub_banks.items():
sort_by = params.sort_by if params.sort_by else self.svd.sort_by
if sort_by is None:
raise ValueError('The sort-by option is missing for %s bank. It needs to be either of "mchirp", "template_duration" or "mu".' % bank_name)
elif params.num_chi_bins and params.num_mu_bins:
raise ValueError('The num-chi-bins and num-mu-bins options cannot both be set for %s bank.' % bank_name)
elif sort_by == "mu" and params.num_chi_bins:
raise ValueError('The num-chi-bins option cannot be set when sort-by is "mu" for %s bank.' % bank_name)
elif sort_by != "mu" and params.num_mu_bins:
raise ValueError('The num-mu-bins option cannot be set when sort-by is not "mu" for %s bank.' % bank_name)
elif not params.num_chi_bins and not params.num_mu_bins:
raise ValueError('Neither of num-chi-bins or num-mu-bins is set in a config file for %s bank. Either one is required.' % bank_name)
elif not self.svd.sort_by:
raise ValueError('The sort-by option is required. It needs to be either of "mchirp", "template_duration" or "mu".')
elif self.svd.num_chi_bins and self.svd.num_mu_bins:
raise ValueError('The num-chi-bins and num-mu-bins options cannot both be set.')
elif self.svd.sort_by == "mu" and self.svd.num_chi_bins:
raise ValueError('The num-chi-bins option cannot be set when sort-by is "mu".')
elif self.svd.sort_by != "mu" and self.svd.num_mu_bins:
raise ValueError('The num-mu-bins option cannot be set when sort-by is not "mu".')
elif not self.svd.num_chi_bins and not self.svd.num_mu_bins:
raise ValueError('Neither of num-chi-bins or num-mu-bins is set in a config file. Either one is required.')
# validate injection settings
if self.filter.injections:
if not self.source.inj_channel_name:
raise ValueError('Must provide injection channel names when config.filter.injections is set.')
if not self.source.inj_shared_memory_dir:
print('WARNING: no inj-shared-memory-dir specified in config, defaulting to use shared-memory-dir.', file=sys.stderr)
self.source.inj_shared_memory_dir = self.source.shared_memory_dir
else:
if self.upload.enable_injection_uploads:
raise ValueError('Cannot provide config.upload.enable_injection_uploads when config.filter.injections is not set.')
[docs] def setup(self):
"""
Set up binning, load relevant analysis files.
"""
super().setup()
# load bin-option file
self.load_svd_options(self.svd.option_file)
# calculate start pad for filtering
max_duration = max(svd_bin["max_dur"] for svd_bin in self.svd.stats.bins.values())
self.filter.start_pad = 16 * self.psd.fft_length + max_duration
# create time bins
if self.span != segment(0, 0):
self.create_time_bins(start_pad=self.filter.start_pad)
[docs] def load_svd_options(self, options_file):
with open(options_file, "r") as f:
svd_stats = dotdict(replace_keys(json.load(f)))
# load default config for sub banks if available
if "sub_banks" in self.svd:
reduced_config = self.svd.copy()
reduced_config.pop("sub_banks")
for sub_bank, props in self.svd.sub_banks.items():
self.svd.sub_banks[sub_bank] = dotdict(replace_keys({**reduced_config, **props}))
# define svd bins, metadata
self.svd.bins = svd_stats.bins.keys()
self.svd.stats = svd_stats
[docs] def write_svd_options(self, output_file):
with open(output_file, "w") as f:
f.write(json.dumps(replace_keys(self.svd.stats, reverse=True), sort_keys = True, indent = 4))