"""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
[docs]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('_', '-')
[docs]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