Source code for pipeparts.pipetools

"""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