gstpipetools module

Miscellaneous utilities for working with Gstreamer Pipelines

References:

[1] 1.0 API: https://lazka.github.io/pgi-docs/Gst-1.0/index.html

gstpipetools.is_element(x: Any) → bool

Test whether an object is a Gst Element. TODO: do these belong somewhere more general?

Args:

x:

Any, the object to test

Returns:

bool, True if x is a Gst Element, false otherwise

gstpipetools.is_pad(x: Any) → bool

Test whether an object is a Gst Pad. TODO: do these belong somewhere more general?

Args:

x:

Any, the object to test

Returns:

bool, True if x is a Gst Pad, false otherwise

gstpipetools.make_element(type_name: str, name: Optional[str] = None, **properties: dict) → Element

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:

type_name:

str, the name of the type of element

elem_name:

str, default None, the name of the element instance

properties:

dict, keyword arguments to set as properties of the element

References:

[1] ElementFactor.make: https://lazka.github.io/pgi-docs/Gst-1.0/classes/ElementFactory.html#Gst.ElementFactory.make

Returns:

Gst.Element

gstpipetools.make_pipeline(name: str) → Pipeline

Create a Pipeline, which is the macroscopic container for Gstreamer elements and is necessary prior to creating any elements

Args:

name:

str, the name of the pipeline

References:

[1] https://gstreamer.freedesktop.org/documentation/application-development/introduction/basics.html?gi-language=c#bins-and-pipelines

Returns:

Pipeline

gstpipetools.run_pipeline(pipeline: Pipeline, segment: Tuple[Union[int, float, LIGOTimeGPS], Union[int, float, LIGOTimeGPS]] = (None, None), handlers: Optional[Iterable[Callable]] = None)

Run a pipeline using the main event loop

Args:

pipeline:

Gst.Pipeline, the pipeline to run

segment:

Tuple[Time, Time], a playback segment

handlers:

Iterable[Callable], default None, an optional list of functions that return Handlers. Each must accept only a “mainloop” and “pipeline” argument.

Returns:

None

gstpipetools.seek(pipeline, segment: ~typing.Tuple[~typing.Union[int, float, ~lal.LIGOTimeGPS], ~typing.Union[int, float, ~lal.LIGOTimeGPS]], flags=<flags GST_SEEK_FLAG_FLUSH of type Gst.SeekFlags>)

Create a new seek event, i.e., Gst.Event.new_seek() for a given gps_start_time and gps_end_time, with optional flags.

Args:

pipeline: gps_start_time:

start time as LIGOTimeGPS, float

gps_end_time:

end time as LIGOTimeGPS, float

flags:

Optional flags, see [2] for options

Notes:

In contrast to the documentation, we set the seek event directly on the pipeline sources. This is because of implementation issues in framecpp demux that prevent the full backward propagation of the seek event from sink -> source (as is done in the gstreamer documentation [1]).

References:

[1] https://gstreamer.freedesktop.org/documentation/additional/design/seeking.html?gi-language=python [2] Flags https://gstreamer.freedesktop.org/documentation/gstreamer/gstsegment.html?gi-language=python#GstSeekFlags

Returns:

None

gstpipetools.seek_args(time: Union[int, float, LIGOTimeGPS]) → Tuple[SeekType, int]

Convenience function for determining the type of arguments to seek for a given time input

Args:

time:

Time, either a float or LIGOTimeGPS

Returns:

Tuple[Gst.SeekType, int]

gstpipetools.to_caps(x: Union[str, Caps]) → Caps

Create a Caps object from a string, or pass thru if already is Caps

Args:

x:

str or Caps, if a string construct Caps instance from x, else pass thru as Caps

Returns:

Caps