# 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.
from ligo.segments import segment
from gstlal.config import Config as BaseConfig
from gstlal.config import dotdict, replace_keys
from gstlal.snax import multichannel_datasource
[docs]
class Config(BaseConfig):
"""
Hold configuration used for SNAX analyzes.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
assert len(self.ifos) == 1, "only one instrument is allowed to be processed at once"
self.ifo = self.ifos[0]
# section-specific options
self.features = dotdict(replace_keys(kwargs["features"]))
self.output = dotdict(replace_keys(kwargs["output"]))
if "metrics" in kwargs:
self.metrics = dotdict(replace_keys(kwargs["metrics"]))
if "services" in kwargs:
self.services = dotdict(replace_keys(kwargs["services"]))
if "stream" in kwargs:
self.stream = dotdict(replace_keys(kwargs["stream"]))
# add convenience properties
self.features.start_pad = 16 * self.psd.fft_length + 30
# create time bins
if self.span != segment(0, 0):
self.create_time_bins(
start_pad=self.features.start_pad,
one_ifo_length=3600
)
# create channel groups
self.channels = multichannel_datasource.channel_dict_from_channel_ini(**self.source)
self.channel_groups = multichannel_datasource.partition_channels_to_equal_subsets(
self.channels,
self.source.get("max_streams", 120),
self.source.get("min_sample_rate", 32),
self.source.get("max_sample_rate", 4096),
)
self.channel_bins = [f"{i:04d}" for i, _ in enumerate(self.channel_groups)]