pipetools.py

"""Utilities for constructing pipeline elements
 
"""
import functools
import inspect
from typing import Union, Optional
 
import gi
import ligo.segments
from lal import LIGOTimeGPS
 
gi.require_version('Gst', '1.0')
from gi.repository import GObject
from gi.repository import Gst
 
from gstlal import gstpipetools
 
GObject.threads_init()
Gst.init(None)
 
PROPERTY_BUILTIN_NAMES = ('async')
 
# The below are used for type annotations
Pipeline = Gst.Pipeline
Element = Gst.Element
Caps = Gst.Caps
GValueArray = Gst.ValueArray # Verify this type
ValueArray = GObject.ValueArray
Segment = ligo.segments.segment
TimeGPS = LIGOTimeGPS
 
 
def clean_property_name(name: str):
    """Utility function for cleaning a property name, handles two cases:
 
        1. Underscores to hyphens: property_name -> property-name
        2. Removes trailing underscores originally used to avoid collisions with builtin: math_ -> math.
            The specific list of builtin collisions detected is defined in PROPERTY_BUILTIN_NAMES
 
    Args:
        name:
            str, the python property name
 
    Returns:
        str, the formatted gstreamer compatible element property name
    """
    if name.endswith('_') and name[:-1] in PROPERTY_BUILTIN_NAMES:
        return name[:-1]
    return name.replace('_', '-')
 
 
def make_element_with_src(pipeline: Gst.Pipeline, src: Union[Gst.Element, Gst.Pad], elem_type_name: str, **properties):
    """Create a new element of the type defined by the given element factory. If name is None, then the
    element will receive a guaranteed unique name, consisting of the element factory name and a number.
    If name is given, it will be given the name supplied.
 
    Args:
        pipeline:
            Pipeline, the existing pipeline to which the new element will be added
        src:
            Element or Pad, the element or pad to use as a source for the new element
        elem_type_name:
            str or None, name of new element, or None to automatically create a unique name
        **properties:
            dict, keyword arguments to be set as properties on the new element
 
    References:
        [1] https://lazka.github.io/pgi-docs/Gst-1.0/classes/ElementFactory.html#Gst.ElementFactory.make
 
    Raises:
        RuntimeError:
            If any problem occurs creating the element.
 
    Returns:
        Element, the new element
    """
    # Determine element name
    name = properties.pop('name') if 'name' in properties else None
 
    properties = {clean_property_name(key): value for key, value in properties.items()}
 
    # Make the element
    elem = gstpipetools.make_element(elem_type_name, name, **properties)
 
    # Add element to pipeline
    pipeline.add(elem)
 
    # Link src argument as input to element
    if gstpipetools.is_pad(src):
        src.get_parent_element().link_pads(src, elem, None)
    elif src is not None:
        src.link(elem)
 
    return elem