"""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
## Adds a <a href="@gstpluginsgooddoc/gst-plugins-good-plugins-audiochebband.html">audiochebband</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstpluginsgooddoc/gst-plugins-good-plugins-audiocheblimit.html">audiocheblimit</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstdoc/gstreamer-plugins-capsfilter.html">capsfilter</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstlalgtkdoc/GSTLALWhiten.html">lal_whiten</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstpluginsgooddoc/gst-plugins-good-plugins-audiofirfilter.html">audiofirfilter</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstlalgtkdoc/GSTLALGate.html">lal_gate</a> element to a pipeline with useful default properties
[docs]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
## Adds a <a href="@gstpluginsgooddoc/gst-plugins-good-plugins-audioiirfilter.html">audioiirfilter</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstlalgtkdoc/GSTLALNoFakeDisconts.html">lal_nofakedisconts</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstlalgtkdoc/GSTLALStateVector.html">lal_statevector</a> element to a pipeline with useful default properties
[docs]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)
## Adds a <a href="@gstlalgtkdoc/GSTLALSimulation.html">lal_simulation</a> element to a pipeline with useful default properties
[docs]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)