trigger.py

"""Module for making trigger elements
 
"""
import numpy
 
from gstlal import pipeio
from gstlal.pipeparts import pipetools
 
 
def trigger(pipeline: pipetools.Pipeline, src: pipetools.Element, n: int, autocorrelation_matrix: numpy.ndarray = None, mask_matrix: numpy.ndarray = None,
            snr_thresh: float = 0, sigmasq: float = None, max_snr: bool = False):
    """Generic trigger generator, find triggers in snr streams
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        n:
            int, number of samples over which to identify triggers
        autocorrelation_matrix:
            ndarray, Array of complex autocorrelation vectors. Number of vectors (rows) in matrix sets number of channels.
            All vectors must have the same length.
        mask_matrix:
            ndarray, Array of integer autocorrelation mask vectors. Number of vectors (rows) in mask sets number of channels.
            All vectors must have the same length. The mask values are either 0 or 1 and indicate whether to use the corresponding
            matrix entry in computing the autocorrelation chi-sq statistic.
        snr_thresh:
            float, default 0, SNR Threshold that determines a trigger
        sigmasq:
            float, default None
        max_snr:
            bool, default False, Set flag to return single loudest trigger from buffer
 
    References:
        [1] GstLAL Trigger Implementation: gstlal/gstlal-burst/gst/lal/gstlal_trigger.c
 
    Returns:
        Element, the trigger element
    """
    properties = {
        "n": n,
        "snr_thresh": snr_thresh,
        "max_snr": max_snr
    }
    if autocorrelation_matrix is not None:
        properties["autocorrelation_matrix"] = pipeio.repack_complex_array_to_real(autocorrelation_matrix)
    if mask_matrix is not None:
        properties["autocorrelation_mask"] = mask_matrix
    if sigmasq is not None:
        properties["sigmasq"] = sigmasq
    return pipetools.make_element_with_src(pipeline, src, "lal_trigger", **properties)
 
 
def burst_trigger_gen(pipeline, src, **properties):
    """Burst Triggergen, find burst triggers in snr streams
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        **properties:
 
    References:
        [1] BurstTriggerGen Implementation: gstlal/gstlal-burst/gst/lal/gstlal_burst_triggergen.c
 
    Returns:
        Element, the trigger gen element
    """
    return pipetools.make_element_with_src(pipeline, src, "lal_bursttriggergen", **properties)
 
 
def blcbc_trigger_gen(pipeline: pipetools.Pipeline, snr: pipetools.Element, chisq: pipetools.Element, template_bank_filename: str, snr_threshold: float,
                      sigmasq: pipetools.ValueArray) -> pipetools.Element:
    """ Produce sngl_inspiral records from SNR and chi squared.
    A trigger is recorded for every instant at which the absolute value of the SNR
    is greater than snr-thresh, and also greater than at all of the window seconds
    of data that come before and after.  snr-thresh and window are properties of
    this element.  This element has a bounded latency of ~ window and is suitable
    for online applications, or applications where precise triggering behavior with
    small chunks of data is required
 
    The maximum possible trigger rate is (1/window) Hz per template.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        snr:
            Gst.Element, the source element
        chisq:
            Element
        template_bank_filename:
            str, Path to XML file used to generate the template bank.  Setting this property resets sigmasq to a vector of 0s.
        snr_threshold:
            float, SNR Threshold that determines a trigger.
        sigmasq:
            Vector of \\sigma^{2} factors.
 
    References:
        [1] gstlal/gstlal-inspiral/gst/lal/gstlal_blcbc_triggergen.c
 
    Returns:
        Element, the blcbc trigger gen
    """
    # snr is complex and chisq is real so the correct source and sink
    # pads will be selected automatically
    elem = pipetools.make_element_with_src(pipeline, snr, "lal_blcbctriggergen", bank_filename=template_bank_filename, snr_thresh=snr_threshold, sigmasq=sigmasq)
    chisq.link(elem)
    return elem
 
 
def trigger_gen(pipeline: pipetools.Pipeline, snr: pipetools.Element, chisq: pipetools.Element, template_bank_filename: str, snr_threshold: float,
                sigmasq: pipetools.ValueArray) -> pipetools.Element:
    """Produce sngl_inspiral records from SNR and chi squared.
    A trigger is recorded for every instant at which the absolute value of the SNR
    is greater than snr-thresh, and also greater than at all of the max_gap seconds
    of data that come before and after.  snr-thresh and max_gap are properties of
    this element
 
    The maximum possible trigger rate is (1/max_gap) Hz per template.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        snr:
            Gst.Element, the source element
        chisq:
            Element
        template_bank_filename:
            str, Path to XML file used to generate the template bank.  Setting this property resets sigmasq to a vector of 0s.
        snr_threshold:
            float, SNR Threshold that determines a trigger.
        sigmasq:
            Vector of \\sigma^{2} factors.
 
    References:
        Implementation: gstlal-inspiral/gst/lal/gstlal_triggergen.c
 
    Returns:
        Element
    """
    # snr is complex and chisq is real so the correct source and sink
    # pads will be selected automatically
    elem = pipetools.make_element_with_src(pipeline, snr, "lal_triggergen", bank_filename=template_bank_filename, snr_thresh=snr_threshold, sigmasq=sigmasq)
    chisq.link(elem)
    return elem
 
 
def itac(pipeline: pipetools.Pipeline, src: pipetools.Element, n: int, bank: str, autocorrelation_matrix: numpy.ndarray = None,
         mask_matrix: numpy.ndarray = None, snr_thresh=0, sigmasq=None) -> pipetools.Element:
    """Find inspiral triggers in snr streams
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        n:
            int, number of samples over which to identify itacs
        bank:
            str, Path to XML file used to generate the template bank.  Setting this property resets sigmasq to a vector of 0s.
        autocorrelation_matrix:
            array, Array of complex autocorrelation vectors.  Number of vectors (rows) in matrix sets number of channels.  All vectors must have the same length.
        mask_matrix:
            array, Array of integer autocorrelation mask vectors.  Number of vectors (rows) in mask sets number of channels.  All vectors must have the same
            length. The mask values are either 0 or 1 and indicate whether to use the corresponding matrix entry in computing the autocorrelation chi-sq statistic.
        snr_thresh:
            float, SNR Threshold that determines a trigger.
        sigmasq:
            array, Vector of \\sigma^{2} factors.  The effective distance of a trigger is \\sqrt{sigma^{2}} / SNR.
 
    References:
        Implementation: gstlal-inspiral/gst/lal/gstlal_itac.c
 
    Returns:
        Element
    """
    properties = {
        "n": n,
        "bank_filename": bank,
        "snr_thresh": snr_thresh
    }
    if autocorrelation_matrix is not None:
        properties["autocorrelation_matrix"] = pipeio.repack_complex_array_to_real(autocorrelation_matrix)
    if mask_matrix is not None:
        properties["autocorrelation_mask"] = mask_matrix
    if sigmasq is not None:
        properties["sigmasq"] = sigmasq
    return pipetools.make_element_with_src(pipeline, src, "lal_itac", **properties)
 
 
def itacac(pipeline, srcs, **properties):
    """Find coincident inspiral triggers in snr streams from multiple detectors
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        srcs:
            Mapping[str, Gst.Element], maps detector strings to Gst.Elements
        **properties
            maps detector strings to pad properties
 
    References:
        Implementation: gstlal-inspiral/gst/lal/gstlal_itacac.c
 
    Returns:
        Element
    """
    elem = pipetools.make_element_with_src(pipeline, None, "lal_itacac")
 
    for idx, (key, src) in enumerate(srcs.items()):
        pad = elem.get_request_pad(f"sink{idx:d}")
        for prop, val in properties[key].items():
            pad.set_property(prop, pipeio.format_property(val))
        src.srcpads[0].link(pad)
 
    return elem