"""Utilities for registering elements to a mixin class that can call elements as methods
Motivation:
1. Inspectability
By making each method statically as opposed to dynamically (from registry list), the full traceback is inspectable
before runtime, allowing for easier development and clearer, more concise readability.
1. Performance
Dynamically setting methods on import is inefficient, though it is clever
"""
import inspect
import types
from typing import List
from gstlal import pipeparts
from gstlal.pipeparts import condition as pipeparts_condition
try:
from gstlal import lloidparts
except ImportError:
lloidparts = None
[docs]class ElementRegistry:
"""Class that defines the pass-through behavior of registered pipeparts elements that are called
as methods of an instance (as opposed to calling the pipeparts functions directly)
All methods of this class are procedurally generated (except _attach_element), by running the following code
from the test_stream module:
print(test_stream.generate_element_mixin())
Notes:
Motivation:
A alternate implementation set attributes dynamically, which though a smaller amount of code, was not
inspectable before runtime (bad). This mixin class may be less fancy, but it allows for inspection
which will improve development workflow (and demystify the pass-through methods of Stream class)
New Elements:
Since registration happens statically (before runtime), adding a new element to the elements registries will not
have an effect unless a similar method is added to this Mixin class. However, thanks to the test suite, adding a
new element will raise a test failure unless this Mixin class is rebuilt.
"""
def _attach_element(self, func: types.FunctionType, *srcs, **kwargs):
"""Utility function for calling the registered element
Args:
func:
*srcs:
**kwargs:
Returns:
RegisteredElementMixin instance (or subclass instance) with newly attached element
"""
head = func(self.pipeline, self.head, *srcs, **kwargs)
cls = type(self)
return cls(
name=self.name,
mainloop=self.mainloop,
pipeline=self.pipeline,
handler=self.handler,
source=self.source,
head=head,
)
[docs] def channelgram(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkchannelgram"""
return self._attach_element(pipeparts.mkchannelgram, *srcs, **kwargs)
[docs] def spectrumplot(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkspectrumplot"""
return self._attach_element(pipeparts.mkspectrumplot, *srcs, **kwargs)
[docs] def histogram(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkhistogram"""
return self._attach_element(pipeparts.mkhistogram, *srcs, **kwargs)
[docs] def uridecodebin(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkuridecodebin"""
return self._attach_element(pipeparts.mkuridecodebin, *srcs, **kwargs)
[docs] def framecppchanneldemux(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkframecppchanneldemux"""
return self._attach_element(pipeparts.mkframecppchanneldemux, *srcs, **kwargs)
[docs] def framecppchannelmux(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkframecppchannelmux_from_list"""
return self._attach_element(pipeparts.mkframecppchannelmux_from_list, *srcs, **kwargs)
[docs] def framecppfilesink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkframecppfilesink"""
return self._attach_element(pipeparts.mkframecppfilesink, *srcs, **kwargs)
[docs] def multifilesink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkmultifilesink"""
return self._attach_element(pipeparts.mkmultifilesink, *srcs, **kwargs)
[docs] def capsfilter(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkcapsfilter"""
return self._attach_element(pipeparts.mkcapsfilter, *srcs, **kwargs)
[docs] def capssetter(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkcapssetter"""
return self._attach_element(pipeparts.mkcapssetter, *srcs, **kwargs)
[docs] def statevector(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkstatevector"""
return self._attach_element(pipeparts.mkstatevector, *srcs, **kwargs)
[docs] def taginject(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktaginject"""
return self._attach_element(pipeparts.mktaginject, *srcs, **kwargs)
[docs] def firfilter(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkfirfilter"""
return self._attach_element(pipeparts.mkfirfilter, *srcs, **kwargs)
[docs] def iirfilter(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkiirfilter"""
return self._attach_element(pipeparts.mkiirfilter, *srcs, **kwargs)
[docs] def shift(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkshift"""
return self._attach_element(pipeparts.mkshift, *srcs, **kwargs)
[docs] def progressreport(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkprogressreport"""
return self._attach_element(pipeparts.mkprogressreport, *srcs, **kwargs)
[docs] def injections(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkinjections"""
return self._attach_element(pipeparts.mkinjections, *srcs, **kwargs)
[docs] def audiochebband(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudiochebband"""
return self._attach_element(pipeparts.mkaudiochebband, *srcs, **kwargs)
[docs] def audiocheblimit(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudiocheblimit"""
return self._attach_element(pipeparts.mkaudiocheblimit, *srcs, **kwargs)
[docs] def audioamplify(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudioamplify"""
return self._attach_element(pipeparts.mkaudioamplify, *srcs, **kwargs)
[docs] def audioundersample(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudioundersample"""
return self._attach_element(pipeparts.mkaudioundersample, *srcs, **kwargs)
[docs] def resample(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkresample"""
return self._attach_element(pipeparts.mkresample, *srcs, **kwargs)
[docs] def interpolator(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkinterpolator"""
return self._attach_element(pipeparts.mkinterpolator, *srcs, **kwargs)
[docs] def whiten(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkwhiten"""
return self._attach_element(pipeparts.mkwhiten, *srcs, **kwargs)
[docs] def integrate(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkintegrate"""
return self._attach_element(pipeparts.mkintegrate, *srcs, **kwargs)
[docs] def tee(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktee"""
return self._attach_element(pipeparts.mktee, *srcs, **kwargs)
[docs] def adder(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkadder"""
return self._attach_element(pipeparts.mkadder, *srcs, **kwargs)
[docs] def multiplier(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkmultiplier"""
return self._attach_element(pipeparts.mkmultiplier, *srcs, **kwargs)
[docs] def queue(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkqueue"""
return self._attach_element(pipeparts.mkqueue, *srcs, **kwargs)
[docs] def drop(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkdrop"""
return self._attach_element(pipeparts.mkdrop, *srcs, **kwargs)
[docs] def nofakedisconts(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mknofakedisconts"""
return self._attach_element(pipeparts.mknofakedisconts, *srcs, **kwargs)
[docs] def firbank(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkfirbank"""
return self._attach_element(pipeparts.mkfirbank, *srcs, **kwargs)
[docs] def tdwhiten(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktdwhiten"""
return self._attach_element(pipeparts.mktdwhiten, *srcs, **kwargs)
[docs] def trim(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktrim"""
return self._attach_element(pipeparts.mktrim, *srcs, **kwargs)
[docs] def mean(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkmean"""
return self._attach_element(pipeparts.mkmean, *srcs, **kwargs)
[docs] def abs(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkabs"""
return self._attach_element(pipeparts.mkabs, *srcs, **kwargs)
[docs] def pow(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkpow"""
return self._attach_element(pipeparts.mkpow, *srcs, **kwargs)
[docs] def reblock(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkreblock"""
return self._attach_element(pipeparts.mkreblock, *srcs, **kwargs)
[docs] def sumsquares(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mksumsquares"""
return self._attach_element(pipeparts.mksumsquares, *srcs, **kwargs)
[docs] def gate(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkgate"""
return self._attach_element(pipeparts.mkgate, *srcs, **kwargs)
[docs] def bitvectorgen(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkbitvectorgen"""
return self._attach_element(pipeparts.mkbitvectorgen, *srcs, **kwargs)
[docs] def matrixmixer(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkmatrixmixer"""
return self._attach_element(pipeparts.mkmatrixmixer, *srcs, **kwargs)
[docs] def togglecomplex(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktogglecomplex"""
return self._attach_element(pipeparts.mktogglecomplex, *srcs, **kwargs)
[docs] def autochisq(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkautochisq"""
return self._attach_element(pipeparts.mkautochisq, *srcs, **kwargs)
[docs] def fakesink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkfakesink"""
return self._attach_element(pipeparts.mkfakesink, *srcs, **kwargs)
[docs] def filesink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkfilesink"""
return self._attach_element(pipeparts.mkfilesink, *srcs, **kwargs)
[docs] def nxydump(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mknxydump"""
return self._attach_element(pipeparts.mknxydump, *srcs, **kwargs)
[docs] def nxydumpsink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mknxydumpsink"""
return self._attach_element(pipeparts.mknxydumpsink, *srcs, **kwargs)
[docs] def nxydumpsinktee(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mknxydumpsinktee"""
return self._attach_element(pipeparts.mknxydumpsinktee, *srcs, **kwargs)
[docs] def triggergen(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktriggergen"""
return self._attach_element(pipeparts.mktriggergen, *srcs, **kwargs)
[docs] def triggerxmlwritersink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktriggerxmlwritersink"""
return self._attach_element(pipeparts.mktriggerxmlwritersink, *srcs, **kwargs)
[docs] def wavenc(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkwavenc"""
return self._attach_element(pipeparts.mkwavenc, *srcs, **kwargs)
[docs] def vorbisenc(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkvorbisenc"""
return self._attach_element(pipeparts.mkvorbisenc, *srcs, **kwargs)
[docs] def colorspace(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkcolorspace"""
return self._attach_element(pipeparts.mkcolorspace, *srcs, **kwargs)
[docs] def theoraenc(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktheoraenc"""
return self._attach_element(pipeparts.mktheoraenc, *srcs, **kwargs)
[docs] def oggmux(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkoggmux"""
return self._attach_element(pipeparts.mkoggmux, *srcs, **kwargs)
[docs] def avimux(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkavimux"""
return self._attach_element(pipeparts.mkavimux, *srcs, **kwargs)
[docs] def audioconvert(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudioconvert"""
return self._attach_element(pipeparts.mkaudioconvert, *srcs, **kwargs)
[docs] def audiorate(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkaudiorate"""
return self._attach_element(pipeparts.mkaudiorate, *srcs, **kwargs)
[docs] def flacenc(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkflacenc"""
return self._attach_element(pipeparts.mkflacenc, *srcs, **kwargs)
[docs] def ogmvideosink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkogmvideosink"""
return self._attach_element(pipeparts.mkogmvideosink, *srcs, **kwargs)
[docs] def videosink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkvideosink"""
return self._attach_element(pipeparts.mkvideosink, *srcs, **kwargs)
[docs] def autoaudiosink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkautoaudiosink"""
return self._attach_element(pipeparts.mkautoaudiosink, *srcs, **kwargs)
[docs] def playbacksink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkplaybacksink"""
return self._attach_element(pipeparts.mkplaybacksink, *srcs, **kwargs)
[docs] def deglitcher(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkdeglitcher"""
return self._attach_element(pipeparts.mkdeglitcher, *srcs, **kwargs)
[docs] def appsink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkappsink"""
return self._attach_element(pipeparts.mkappsink, *srcs, **kwargs)
[docs] def checktimestamps(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkchecktimestamps"""
return self._attach_element(pipeparts.mkchecktimestamps, *srcs, **kwargs)
[docs] def peak(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkpeak"""
return self._attach_element(pipeparts.mkpeak, *srcs, **kwargs)
[docs] def itac(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkitac"""
return self._attach_element(pipeparts.mkitac, *srcs, **kwargs)
[docs] def itacac(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkitacac"""
return self._attach_element(pipeparts.mkitacac, *srcs, **kwargs)
[docs] def trigger(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktrigger"""
return self._attach_element(pipeparts.mktrigger, *srcs, **kwargs)
[docs] def latency(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mklatency"""
return self._attach_element(pipeparts.mklatency, *srcs, **kwargs)
[docs] def computegamma(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkcomputegamma"""
return self._attach_element(pipeparts.mkcomputegamma, *srcs, **kwargs)
[docs] def bursttriggergen(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkbursttriggergen"""
return self._attach_element(pipeparts.mkbursttriggergen, *srcs, **kwargs)
[docs] def odctodqv(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mkodctodqv"""
return self._attach_element(pipeparts.mkodctodqv, *srcs, **kwargs)
[docs] def tcpserversink(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts.mktcpserversink"""
return self._attach_element(pipeparts.mktcpserversink, *srcs, **kwargs)
[docs] def htgate(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts_condition.mkhtgate"""
return self._attach_element(pipeparts_condition.mkhtgate, *srcs, **kwargs)
[docs] def condition(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts_condition.mkcondition"""
return self._attach_element(pipeparts_condition.mkcondition, *srcs, **kwargs)
[docs] def multiband(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function pipeparts_condition.mkmultiband"""
return self._attach_element(pipeparts_condition.mkmultiband, *srcs, **kwargs)
[docs] def controlsnksrc(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function lloidparts.mkcontrolsnksrc"""
return self._attach_element(lloidparts.mkcontrolsnksrc, *srcs, **kwargs)
[docs] def lloid_branch(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function lloidparts.mkLLOIDbranch"""
return self._attach_element(lloidparts.mkLLOIDbranch, *srcs, **kwargs)
[docs] def lloid_snr_slices_to_timeslice_chisq(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function lloidparts.mkLLOIDSnrSlicesToTimeSliceChisq"""
return self._attach_element(lloidparts.mkLLOIDSnrSlicesToTimeSliceChisq, *srcs, **kwargs)
[docs] def lloid_snr_chisq_to_triggers(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function lloidparts.mkLLOIDSnrChisqToTriggers"""
return self._attach_element(lloidparts.mkLLOIDSnrChisqToTriggers, *srcs, **kwargs)
[docs] def lloid_hoft_to_snr_slices(self, *srcs, **kwargs):
"""Automatically generated pass-through method for function lloidparts.mkLLOIDhoftToSnrSlices"""
return self._attach_element(lloidparts.mkLLOIDhoftToSnrSlices, *srcs, **kwargs)
####################################################################################################
# #
# ELEMENT REGISTRATION TOOLS BELOW #
# #
# Note: this registry is only used in the test suite to verify that the above ElementRegistry #
# has all the expected methods with the desired names #
# #
####################################################################################################
[docs]def module_name(func: types.FunctionType) -> str:
"""Get a functions module name
Args:
func:
Function, the function for which to find module name
Returns:
str, the name of the module where the function is defined
"""
names = inspect.getmodule(func).__name__.split('.')
if names[-1] == '__init__.py': # booo code defined in __init__.py....
return names[-2]
return names[-1]
[docs]class RegisteredElement:
"""A function to register as a method"""
LEGACY_PREFIX = 'mk'
def __init__(self, func: types.FunctionType, base_name: str = None, mod_name: str = None):
self.func = func
self.base_name = func.__name__ if base_name is None else base_name
self.module_name = module_name(self.func) if mod_name is None else mod_name
def __repr__(self):
"""Repr for nice test formatting"""
return 'RegisteredElement({}.{})'.format(self.module_name, self.func.__name__)
@property
def name(self):
"""Method that can be overridden by subclasses for special naming behavior"""
formatted_name = self.base_name
if formatted_name.startswith(self.LEGACY_PREFIX):
formatted_name = formatted_name.replace(self.LEGACY_PREFIX, '')
return formatted_name.lower()
[docs]def generate_registered_method_source(elements: List[RegisteredElement] = None) -> str:
"""This function is only to be used when making changes to the element registry, though it is also
allowed to edit the ElementRegistry class directly. This function helps automatically generate the method
source code for the ElementRegistry class using the attributes of the registered elements.
Args:
elements:
List[RegisteredElement], default None, if None use ALL_ELEMENTS registered below
Returns:
str
"""
if elements is None:
elements = ALL_ELEMENTS
return '\n\n'.join(e.format_method_source() for e in elements)
PIPEPARTS_ELEMENTS = [
RegisteredElement(pipeparts.mkchannelgram),
RegisteredElement(pipeparts.mkspectrumplot),
RegisteredElement(pipeparts.mkhistogram),
RegisteredElement(pipeparts.mkuridecodebin),
RegisteredElement(pipeparts.mkframecppchanneldemux),
RegisteredElement(pipeparts.mkframecppchannelmux_from_list, base_name='mkframecppchannelmux'),
RegisteredElement(pipeparts.mkframecppfilesink),
RegisteredElement(pipeparts.mkmultifilesink),
RegisteredElement(pipeparts.mkcapsfilter),
RegisteredElement(pipeparts.mkcapssetter),
RegisteredElement(pipeparts.mkstatevector),
RegisteredElement(pipeparts.mktaginject),
RegisteredElement(pipeparts.mkfirfilter),
RegisteredElement(pipeparts.mkiirfilter),
RegisteredElement(pipeparts.mkshift),
RegisteredElement(pipeparts.mkprogressreport),
RegisteredElement(pipeparts.mkinjections),
RegisteredElement(pipeparts.mkaudiochebband),
RegisteredElement(pipeparts.mkaudiocheblimit),
RegisteredElement(pipeparts.mkaudioamplify),
RegisteredElement(pipeparts.mkaudioundersample),
RegisteredElement(pipeparts.mkresample),
RegisteredElement(pipeparts.mkinterpolator),
RegisteredElement(pipeparts.mkwhiten),
RegisteredElement(pipeparts.mkintegrate),
RegisteredElement(pipeparts.mktee),
RegisteredElement(pipeparts.mkadder),
RegisteredElement(pipeparts.mkmultiplier),
RegisteredElement(pipeparts.mkqueue),
RegisteredElement(pipeparts.mkdrop),
RegisteredElement(pipeparts.mknofakedisconts),
RegisteredElement(pipeparts.mkfirbank),
RegisteredElement(pipeparts.mktdwhiten),
RegisteredElement(pipeparts.mktrim),
RegisteredElement(pipeparts.mkmean),
RegisteredElement(pipeparts.mkabs),
RegisteredElement(pipeparts.mkpow),
RegisteredElement(pipeparts.mkreblock),
RegisteredElement(pipeparts.mksumsquares),
RegisteredElement(pipeparts.mkgate),
RegisteredElement(pipeparts.mkbitvectorgen),
RegisteredElement(pipeparts.mkmatrixmixer),
RegisteredElement(pipeparts.mktogglecomplex),
RegisteredElement(pipeparts.mkautochisq),
RegisteredElement(pipeparts.mkfakesink),
RegisteredElement(pipeparts.mkfilesink),
RegisteredElement(pipeparts.mknxydump),
RegisteredElement(pipeparts.mknxydumpsink),
RegisteredElement(pipeparts.mknxydumpsinktee),
RegisteredElement(pipeparts.mktriggergen),
RegisteredElement(pipeparts.mktriggerxmlwritersink),
RegisteredElement(pipeparts.mkwavenc),
RegisteredElement(pipeparts.mkvorbisenc),
RegisteredElement(pipeparts.mkcolorspace),
RegisteredElement(pipeparts.mktheoraenc),
RegisteredElement(pipeparts.mkoggmux),
RegisteredElement(pipeparts.mkavimux),
RegisteredElement(pipeparts.mkaudioconvert),
RegisteredElement(pipeparts.mkaudiorate),
RegisteredElement(pipeparts.mkflacenc),
RegisteredElement(pipeparts.mkogmvideosink),
RegisteredElement(pipeparts.mkvideosink),
RegisteredElement(pipeparts.mkautoaudiosink),
RegisteredElement(pipeparts.mkplaybacksink),
RegisteredElement(pipeparts.mkdeglitcher),
RegisteredElement(pipeparts.mkappsink),
RegisteredElement(pipeparts.mkchecktimestamps),
RegisteredElement(pipeparts.mkpeak),
RegisteredElement(pipeparts.mkitac),
RegisteredElement(pipeparts.mkitacac),
RegisteredElement(pipeparts.mktrigger),
RegisteredElement(pipeparts.mklatency),
RegisteredElement(pipeparts.mkcomputegamma),
RegisteredElement(pipeparts.mkbursttriggergen),
RegisteredElement(pipeparts.mkodctodqv),
RegisteredElement(pipeparts.mktcpserversink),
]
CONDITION_ELEMENTS = [
# In the below, we have imported condition module as 'pipeparts_condition' to avoid
# a name collision with the element 'mkcondition', whose method-name will become 'condition'
RegisteredElement(pipeparts_condition.mkhtgate, mod_name='pipeparts_condition'),
RegisteredElement(pipeparts_condition.mkcondition, mod_name='pipeparts_condition'),
RegisteredElement(pipeparts_condition.mkmultiband, mod_name='pipeparts_condition'),
]
if lloidparts is not None:
LLOIDPARTS_ELEMENTS = [
RegisteredElement(lloidparts.mkcontrolsnksrc),
RegisteredElement(lloidparts.mkLLOIDbranch, base_name='mklloid_branch'),
RegisteredElement(lloidparts.mkLLOIDSnrSlicesToTimeSliceChisq, base_name='mklloid_snr_slices_to_timeslice_chisq'),
RegisteredElement(lloidparts.mkLLOIDSnrChisqToTriggers, base_name='mklloid_snrchisq_to_triggers'),
RegisteredElement(lloidparts.mkLLOIDhoftToSnrSlices, base_name='mklloid_hoft_to_snr_slices'),
]
else:
LLOIDPARTS_ELEMENTS = []
ALL_ELEMENTS = PIPEPARTS_ELEMENTS + CONDITION_ELEMENTS + LLOIDPARTS_ELEMENTS