filters.py

"""Module for filter elements
 
"""
from typing import Union
 
import gi
 
gi.require_version('Gst', '1.0')
from gi.repository import GObject
from gi.repository import Gst
 
GObject.threads_init()
Gst.init(None)
 
from gstlal import gstpipetools
from gstlal.pipeparts import pipetools
 
 

def audio_cheb_band(pipeline: pipetools.Pipeline, src: pipetools.Element, lower_frequency: float, upper_frequency: float, poles: int = 8) -> pipetools.Element:
    """Attenuates all frequencies outside (bandpass) or inside (bandreject) of a frequency band. The number
    of poles and the ripple parameter control the rolloff. This element has the advantage over the windowed
    sinc bandpass and bandreject filter that it is much faster and produces almost as good results. It's only
    disadvantages are the highly non-linear phase and the slower rolloff compared to a windowed sinc filter
    with a large kernel. For type 1 the ripple parameter specifies how much ripple in dB is allowed in the
    passband, i.e. some frequencies in the passband will be amplified by that value. A higher ripple value
    will allow a faster rolloff. For type 2 the ripple parameter specifies the stopband attenuation. In the
    stopband the gain will be at most this value. A lower ripple value will allow a faster rolloff. As a
    special case, a Chebyshev type 1 filter with no ripple is a Butterworth filter.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        lower_frequency:
            float, Start frequency of the band (Hz)
        upper_frequency:
            float, Stop frequency of the band (Hz)
        poles:
            int, Number of poles to use, will be rounded up to the next multiple of four
 
    References:
        [1] https://gstreamer.freedesktop.org/documentation/audiofx/audiochebband.html?gi-language=python
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "audiochebband", lower_frequency=lower_frequency, upper_frequency=upper_frequency, poles=poles)
 
 

def audio_cheb_limit(pipeline: pipetools.Pipeline, src: pipetools.Element, cutoff: float, mode: int = 0, poles: int = 8, type: int = 1, ripple: float = 0.25) -> pipetools.Element:
    """Attenuates all frequencies above the cutoff frequency (low-pass) or all frequencies below the cutoff frequency (high-pass).
    The number of poles and the ripple parameter control the rolloff. This element has the advantage over the windowed sinc lowpass
    and highpass filter that it is much faster and produces almost as good results. It's only disadvantages are the highly non-linear
    phase and the slower rolloff compared to a windowed sinc filter with a large kernel. For type 1 the ripple parameter specifies
    how much ripple in dB is allowed in the passband, i.e. some frequencies in the passband will be amplified by that value. A higher
    ripple value will allow a faster rolloff. For type 2 the ripple parameter specifies the stopband attenuation. In the stopband the
    gain will be at most this value. A lower ripple value will allow a faster rolloff. As a special case, a Chebyshev type 1 filter
    with no ripple is a Butterworth filter.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        cutoff:
            float, Cut off frequency (Hz)
        mode:
            int, default 0, 0 for low-pass or 1 for high-pass
        poles:
            int, Number of poles to use, will be rounded up to the next multiple of four
        type:
            int, default 1, Type of the chebychev filter
        ripple:
            float, default 0,25, Amount of ripple (dB)
 
    References:
        [1] https://gstreamer.freedesktop.org/documentation/audiofx/audiocheblimit.html?gi-language=python
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "audiocheblimit", cutoff=cutoff, mode=mode, poles=poles, type=type, ripple=ripple)
 
 

def caps(pipeline: pipetools.Pipeline, src: pipetools.Element, caps: Union[str, pipetools.Caps], **properties: dict) -> pipetools.Element:
    """The element does not modify data as such, but can enforce limitations on the data format.
    Note: this element does *not* act as a filter on the data of the source, but rather as a filter on the
    metadata (CAPS) of the source element.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        caps:
            str or Gst.Caps, the capabilities specification to limit the source data format
        **properties:
            dict, keyword arguments to be set as element properties
 
    References:
        [1] capsfilter docs: https://gstreamer.freedesktop.org/documentation/coreelements/capsfilter.html?gi-language=python
 
    Returns:
        Element, the source element limited by the given caps (capabilities)
    """
    return pipetools.make_element_with_src(pipeline, src, "capsfilter", caps=gstpipetools.to_caps(caps), **properties)
 
 

def drop(pipeline: pipetools.Pipeline, src: pipetools.Element, drop_samples: int = 0) -> pipetools.Element:
    """Drop samples from the start of a stream
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        srcs:
            Iterable[Gst.Element], the source elements
        drop_samples:
            int, default 0, number of samples to drop from the beginning of a stream
 
    References:
        Implementation: gstlal/gst/lal/gstlal_drop.c
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "lal_drop", drop_samples=drop_samples)
 
 

def fir(pipeline: pipetools.Pipeline, src: pipetools.Element, kernel: pipetools.GValueArray, latency, **properties: dict) -> pipetools.Element:
    """Generic audio FIR filter. Before usage the "kernel" property has to be set to the filter kernel that should be
    used and the "latency" property has to be set to the latency (in samples) that is introduced by the filter kernel.
    Setting a latency of n samples will lead to the first n samples being dropped from the output and n samples added
    to the end.
 
    The filter kernel describes the impulse response of the filter. To calculate the frequency response of the filter
    you have to calculate the Fourier Transform of the impulse response.
 
    To change the filter kernel whenever the sampling rate changes the "rate-changed" signal can be used. This should
    be done for most FIR filters as they're depending on the sampling rate.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        kernel:
            Gst.GValueArray, filter kernel for the FIR filter
        latency:
            int, filter latency in samples
        **properties:
            dict, keyword arguments to be set as element properties
 
    References:
        [1] audiofirfilter docs: https://gstreamer.freedesktop.org/documentation/audiofx/audiofirfilter.html?gi-language=python
 
    Returns:
        Element, the FIR element
    """
    properties.update((name, val) for name, val in (("kernel", kernel), ("latency", latency)) if val is not None)
    return pipetools.make_element_with_src(pipeline, src, "audiofirfilter", **properties)
 
 

def gate(pipeline: pipetools.Pipeline, src: pipetools.Element, threshold: float = None, control: pipetools.Element = None, **properties) -> pipetools.Element:
    """Flag buffers as gaps based on the value of a control input
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        threshold:
            float, default None, uutput will be flagged as non-gap when magnitude of control input is >= this value.  See also invert-control.
        control:
            Element, optional control element input
        **properties:
            emit_signals:
                bool, Emit start and stop signals (rate-changed is always emited).  The start and stop signals
                are emited on gap-to-non-gap and non-gap-to-gap transitions in the output stream respectively.
 
    References:
        Implementation: gstlal/gst/lal/gstlal_gate.c
 
    Returns:
        Element
    """
    if threshold is not None:
        elem = pipetools.make_element_with_src(pipeline, None, "lal_gate", threshold=threshold, **properties)
    else:
        elem = pipetools.make_element_with_src(pipeline, None, "lal_gate", **properties)
    for peer, padname in ((src, "sink"), (control, "control")):
        if isinstance(peer, Gst.Pad):
            peer.get_parent_element().link_pads(peer, elem, padname)
        elif peer is not None:
            peer.link_pads(None, elem, padname)
    return elem
 
 

def iir(pipeline: pipetools.Pipeline, src: pipetools.Element, a: pipetools.ValueArray, b: pipetools.ValueArray) -> pipetools.Element:
    """aGeneric audio IIR filter. Before usage the "a" and "b" properties have to be set to the filter coefficients
    that should be used.
 
    The filter coefficients describe the numerator and denominator of the transfer function.
 
    To change the filter coefficients whenever the sampling rate changes the "rate-changed" signal can be used.
    This should be done for most IIR filters as they're depending on the sampling rate.
 
        convention is z = \exp(-i 2 \pi f / f_{\rm sampling})
        H(z) = (\sum_{j=0}^{N} a_j z^{-j}) / (\sum_{j=0}^{N} (-1)^{j} b_j z^{-j})
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        a:
            ValueArray, Filter coefficients (denominator of transfer function)
        b:
            ValueArray, Filter coefficients (numerator of transfer function)
 
    References:
        [1] audioiirfilter docs: https://gstreamer.freedesktop.org/documentation/audiofx/audioiirfilter.html?gi-language=python
 
    Returns:
        Element, IIR of the sources
    """
    #
    return pipetools.make_element_with_src(pipeline, src, "audioiirfilter", a=a, b=b)
 
 

def remove_fake_disconts(pipeline: pipetools.Pipeline, src: pipetools.Element, silent: bool = True) -> pipetools.Element:
    """Fix incorrectly-set discontinuity flags
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        srcs:
            Iterable[Gst.Element], the source elements
        silent:
            bool, default True, if True Don't print a message when alterning the flags in a buffer.
 
    References:
        Implementation: gstal/gst/lal/gstlal_nofakedisconts.c
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "lal_nofakedisconts", silent=silent)
 
 

def state_vector(pipeline: pipetools.Pipeline, src: pipetools.Element, **properties) -> pipetools.Element:
    """Converts a state vector stream into booleans, for example to drive a lal_gate element.
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        **properties:
 
    References:
        Implementation: gstlal/gst/lal/gstlal_statevector.c
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "lal_statevector", **properties)
 
 

def inject(pipeline: pipetools.Pipeline, src: pipetools.Element, filename: str) -> pipetools.Element:
    """An injection routine calling lalsimulation waveform generators
 
    Args:
        pipeline:
            Gst.Pipeline, the pipeline to which the new element will be added
        src:
            Gst.Element, the source element
        filename:
            str, path to xml file Name of LIGO Light Weight XML file containing list(s) of software injections
 
    References:
        Implementation: gstlal/gst/lal/gstlal_simulation.c
 
    Returns:
        Element
    """
    return pipetools.make_element_with_src(pipeline, src, "lal_simulation", xml_location=filename)