pipeline

Frame pipelines: apply a chain of stages to a stream of frames in one pass.

A pipeline reads frames from one or more GWF files, passes them through a sequence of :class:Stage objects, and writes the result. Every stage is a transformation of an iterator of :class:~gwframe.write.Frame objects, so stages compose freely and the whole chain costs a single decode/encode pass regardless of its length.

Two output modes are supported:

• per file (the default): each input file is processed independently and written under its own name in the output directory, exactly like the single operations in :mod:gwframe.operations.
• re-chunked (file_duration given): the inputs are read in GPS order as one continuous stream and the output is cut into files of the requested duration, named PREFIX-GPS-DURATION.gwf.

Examples:

>>> from gwframe.pipeline import Pipeline, RenameStage, DropStage, ResizeStage
>>> pipeline = Pipeline([
...     RenameStage({"L1:GDS-CALIB_STRAIN": "L1:STRAIN"}),
...     DropStage(["L1:DEBUG"]),
...     ResizeStage(4.0),
... ])
>>> pipeline.run(["raw/a.gwf", "raw/b.gwf"], "curated/")

DropStage

DropStage(channels: Sequence[str])

Bases: MapStage

Remove channels.

Parameters:

Name	Type	Description	Default
channels	sequence of str	Channel names to remove. Channels absent from a frame are skipped.	required

Source code in gwframe/pipeline.py

def __init__(self, channels: Sequence[str]):
    if not channels:
        msg = "channels must be provided and non-empty"
        raise ValueError(msg)
    self.channels = list(dict.fromkeys(channels))

FillGapsStage

FillGapsStage(fill_value: float = 0.0, *, invalid: bool = False)

Bases: Stage

Fill gaps in GPS time between consecutive frames with synthetic frames.

Wherever a frame starts later than the previous one ended, frames of a constant fill value are inserted to cover the gap, so the stream (and files re-chunked from it) is contiguous. The inserted frames copy the previous frame's channels, sample rates, dtypes, units and duration; the last inserted frame is shorter if the gap is not a whole multiple of that duration. Nothing is inserted before the first frame. Frames that overlap the previous one are an error.

This mirrors GStreamer's audiorate skip-to-first=true element as used in gstlal frame-rewriting pipelines.

Parameters:

Name	Type	Description	Default
fill_value	float	Sample value for the inserted frames, cast to each channel's dtype (default: 0.0)	0.0
invalid	bool	If True, flag the inserted data as invalid so consumers can tell it from real data: ADC channels get their channel-level dataValid flag set (they read back fully masked). Proc and sim channels have no such flag in the frame format and are left unflagged; a warning names them once.	False

Source code in gwframe/pipeline.py

def __init__(self, fill_value: float = 0.0, *, invalid: bool = False):
    self.fill_value = fill_value
    self.invalid = invalid
    self._unflagged_warned = False

ImputeStage

ImputeStage(replace_value: float = nan, fill_value: float = 0.0, channels: Sequence[str] | None = None)

Bases: MapStage

Replace a value in channel data with a fill value.

Parameters:

Name	Type	Description	Default
replace_value	float	Value to replace (default: NaN)	nan
fill_value	float	Replacement, cast to each channel's dtype (default: 0.0)	0.0
channels	sequence of str	Restrict to these channels (default: all channels)	None

Source code in gwframe/pipeline.py

def __init__(
    self,
    replace_value: float = np.nan,
    fill_value: float = 0.0,
    channels: Sequence[str] | None = None,
):
    self.replace_value = replace_value
    self.fill_value = fill_value
    self.channels = list(dict.fromkeys(channels)) if channels else None

MapStage

Bases: Stage

A stage that transforms frames one at a time.

transform abstractmethod

transform(frame: Frame) -> Frame

Transform a single frame, returning it (or a replacement).

Source code in gwframe/pipeline.py

@abstractmethod
def transform(self, frame: Frame) -> Frame:
    """Transform a single frame, returning it (or a replacement)."""

Pipeline

Pipeline(stages: Sequence[Stage] = ())

A sequence of stages applied to a stream of frames.

Parameters:

Name	Type	Description	Default
stages	sequence of Stage	Stages to apply, in order. May be empty, in which case the pipeline simply rewrites its input (useful for recompressing or re-chunking).	()

Source code in gwframe/pipeline.py

def __init__(self, stages: Sequence[Stage] = ()):
    self.stages = list(stages)

apply

apply(frames: Iterator[Frame]) -> Iterator[Frame]

Apply every stage, in order, to a stream of frames.

Source code in gwframe/pipeline.py

def apply(self, frames: Iterator[Frame]) -> Iterator[Frame]:
    """Apply every stage, in order, to a stream of frames."""
    for stage in self.stages:
        frames = stage.apply(frames)
    return frames

preflight

preflight(channels: Iterable[str]) -> tuple[set[str], list[str]]

Check the pipeline against a channel set without reading any data.

Parameters:

Name	Type	Description	Default
channels	iterable of str	Channel names of the input, typically from :func:gwframe.inspect.get_channels on the first file	required

Returns:

Name	Type	Description
channels	set[str]	Channel names expected in the output
warnings	list[str]	Likely mistakes, one per line, each prefixed with the stage that raised it and the point in the chain the check was made against (the input, or the output of the previous stage)

Raises:

Type	Description
ValueError	A stage cannot be applied, or no channels would remain.

Source code in gwframe/pipeline.py

def preflight(self, channels: Iterable[str]) -> tuple[set[str], list[str]]:
    """
    Check the pipeline against a channel set without reading any data.

    Parameters
    ----------
    channels : iterable of str
        Channel names of the input, typically from
        :func:`gwframe.inspect.get_channels` on the first file

    Returns
    -------
    channels : set[str]
        Channel names expected in the output
    warnings : list[str]
        Likely mistakes, one per line, each prefixed with the stage
        that raised it and the point in the chain the check was made
        against (the input, or the output of the previous stage)

    Raises
    ------
    ValueError
        A stage cannot be applied, or no channels would remain.
    """
    current = set(channels)
    warnings: list[str] = []
    where = "in the input"
    for index, stage in enumerate(self.stages, start=1):
        label = f"stage {index} ({stage.name})"
        try:
            current, stage_warnings = stage.project(current)
        except ValueError as e:
            msg = f"{label}: {e}"
            raise ValueError(msg) from e
        warnings.extend(f"{label}: {where}, {w}" for w in stage_warnings)
        where = f"after {label}"
        if not current:
            msg = f"{label}: no channels would remain"
            raise ValueError(msg)
    return current, warnings

run

run(input_files: str | PathLike[str] | Sequence[str | PathLike[str]], output_dir: str | PathLike[str], *, compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6, file_duration: float | None = None, prefix: str | None = None, progress: ProgressCallback | None = None, strict: bool = False) -> list[str]

Run the pipeline over files, writing the result.

Parameters:

Name	Type	Description	Default
input_files	str, path-like, or sequence of str/path-like	Input GWF file(s)	required
output_dir	str or path - like	Directory where output files are written (created if needed)	required
compression	int	Compression scheme for the output	ZERO_SUPPRESS_OTHERWISE_GZIP
compression_level	int	Compression level 0-9 (default: 6)	6
file_duration	float	If given, read the inputs in GPS order as one stream and cut the output into files of at most this many seconds (a gap in GPS time also starts a new file). Otherwise each input file is written under its own name.	None
prefix	str	OBSERVATORY-DESCRIPTION prefix for re-chunked file names (default: taken from the first input file). Ignored without file_duration.	None
progress	callable	Called after each input frame has been processed with (frames_done, frames_total, path), where path is the input file the frame came from, for progress reporting	None
strict	bool	If True, :meth:preflight warnings (such as a channel a stage names but the input lacks) are errors and nothing is written	False

Returns:

Name	Type	Description
output_files	list[str]	Output file paths, in the order written

Raises:

Type	Description
ValueError	The pipeline cannot be applied to the input (see :meth:preflight), a stage rejects the data, or strict is set and preflight raised warnings.

Source code in gwframe/pipeline.py

def run(
    self,
    input_files: str | PathLike[str] | Sequence[str | PathLike[str]],
    output_dir: str | PathLike[str],
    *,
    compression: int = Compression.ZERO_SUPPRESS_OTHERWISE_GZIP,
    compression_level: int = 6,
    file_duration: float | None = None,
    prefix: str | None = None,
    progress: ProgressCallback | None = None,
    strict: bool = False,
) -> list[str]:
    """
    Run the pipeline over files, writing the result.

    Parameters
    ----------
    input_files : str, path-like, or sequence of str/path-like
        Input GWF file(s)
    output_dir : str or path-like
        Directory where output files are written (created if needed)
    compression : int, optional
        Compression scheme for the output
    compression_level : int, optional
        Compression level 0-9 (default: 6)
    file_duration : float, optional
        If given, read the inputs in GPS order as one stream and cut the
        output into files of at most this many seconds (a gap in GPS
        time also starts a new file). Otherwise each input file is
        written under its own name.
    prefix : str, optional
        ``OBSERVATORY-DESCRIPTION`` prefix for re-chunked file names
        (default: taken from the first input file). Ignored without
        ``file_duration``.
    progress : callable, optional
        Called after each input frame has been processed with
        ``(frames_done, frames_total, path)``, where ``path`` is the
        input file the frame came from, for progress reporting
    strict : bool, optional
        If True, :meth:`preflight` warnings (such as a channel a stage
        names but the input lacks) are errors and nothing is written

    Returns
    -------
    output_files : list[str]
        Output file paths, in the order written

    Raises
    ------
    ValueError
        The pipeline cannot be applied to the input (see
        :meth:`preflight`), a stage rejects the data, or ``strict`` is
        set and preflight raised warnings.
    """
    if isinstance(input_files, str | PathLike):
        input_files = [input_files]
    files = [Path(f) for f in input_files]
    if not files:
        msg = "no input files"
        raise ValueError(msg)

    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # One header read per file gives the frame spec, the channel list for
    # preflight, the GPS order and the total frame count for progress
    infos = {f: get_info(f) for f in files}
    ticker = _ProgressTicker(
        progress, sum(info.num_frames for info in infos.values())
    )

    _, warnings = self.preflight(infos[files[0]].channels)
    if strict and warnings:
        msg = "preflight warnings (strict): " + "; ".join(warnings)
        raise ValueError(msg)

    if file_duration is None:
        return [
            self._write_file(
                f, infos[f], output_dir, compression, compression_level, ticker
            )
            for f in files
        ]

    if file_duration <= 0:
        msg = "file_duration must be a positive number"
        raise ValueError(msg)
    files = sorted(
        files,
        key=lambda f: (
            infos[f].frames[0].start if infos[f].frames else math.inf,
            f.name,
        ),
    )
    return self._write_chunked(
        files,
        infos[files[0]].frame_spec,
        output_dir,
        compression,
        compression_level,
        file_duration,
        prefix if prefix is not None else file_prefix(files[0]),
        ticker,
    )

RenameStage

RenameStage(channel_map: Mapping[str, str])

Bases: MapStage

Rename channels.

Parameters:

Name	Type	Description	Default
channel_map	mapping of str to str	Old channel name to new channel name. Channels absent from a frame are skipped.	required

Source code in gwframe/pipeline.py

def __init__(self, channel_map: Mapping[str, str]):
    if not channel_map:
        msg = "channel_map must be provided and non-empty"
        raise ValueError(msg)
    self.channel_map = dict(channel_map)
    targets = list(self.channel_map.values())
    duplicates = sorted({t for t in targets if targets.count(t) > 1})
    if duplicates:
        msg = f"multiple channels renamed to the same name: {', '.join(duplicates)}"
        raise ValueError(msg)

ResizeStage

ResizeStage(duration: float)

Bases: Stage

Resize frames to a target duration.

Frames longer than the target are split into equal parts; the source duration must be a whole multiple of the target. Consecutive shorter frames are merged, also across input file boundaries, until they reach the target; a gap in GPS time or the end of the stream flushes a shorter frame. Output frames are numbered sequentially from 0.

Parameters:

Name	Type	Description	Default
duration	float	Target frame duration in seconds	required

Source code in gwframe/pipeline.py

def __init__(self, duration: float):
    if duration is None or duration <= 0:
        msg = "duration must be a positive number"
        raise ValueError(msg)
    self.duration = float(duration)

SelectStage

SelectStage(channels: Sequence[str])

Bases: MapStage

Keep only the given channels, removing all others.

Parameters:

Name	Type	Description	Default
channels	sequence of str	Channel names to keep	required

Source code in gwframe/pipeline.py

def __init__(self, channels: Sequence[str]):
    if not channels:
        msg = "channels must be provided and non-empty"
        raise ValueError(msg)
    self.channels = list(dict.fromkeys(channels))

Stage

Bases: ABC

One step of a frame pipeline.

A stage transforms an iterator of frames into another iterator of frames. Most stages act on one frame at a time; subclass :class:MapStage for those. Stages that change the number of frames (such as :class:ResizeStage) override :meth:apply directly.

Stages can also predict their effect on the set of channel names, which lets a pipeline be checked against a file's table of contents before any data is read (see :meth:Pipeline.preflight).

apply abstractmethod

apply(frames: Iterator[Frame]) -> Iterator[Frame]

Transform a stream of frames.

Source code in gwframe/pipeline.py

@abstractmethod
def apply(self, frames: Iterator[Frame]) -> Iterator[Frame]:
    """Transform a stream of frames."""

project

project(channels: set[str]) -> tuple[set[str], list[str]]

Predict the channel set after this stage.

Parameters:

Name	Type	Description	Default
channels	set[str]	Channel names entering the stage	required

Returns:

Name	Type	Description
channels	set[str]	Channel names leaving the stage
warnings	list[str]	Human-readable notes about likely mistakes, such as a channel that is named by the stage but not present in the input

Raises:

Type	Description
ValueError	The stage cannot be applied to this channel set (for example a rename that would overwrite an existing channel).

Source code in gwframe/pipeline.py

def project(self, channels: set[str]) -> tuple[set[str], list[str]]:
    """
    Predict the channel set after this stage.

    Parameters
    ----------
    channels : set[str]
        Channel names entering the stage

    Returns
    -------
    channels : set[str]
        Channel names leaving the stage
    warnings : list[str]
        Human-readable notes about likely mistakes, such as a channel
        that is named by the stage but not present in the input

    Raises
    ------
    ValueError
        The stage cannot be applied to this channel set (for example a
        rename that would overwrite an existing channel).
    """
    return set(channels), []

file_prefix

file_prefix(path: str | PathLike[str]) -> str

Return the OBSERVATORY-DESCRIPTION prefix of a frame file name.

Files that don't follow the OBS-DESC-GPS-DURATION.gwf convention use their whole stem as the prefix.

Examples:

>>> file_prefix("L-L1_HOFT_C00-1238166018-4096.gwf")
'L-L1_HOFT_C00'
>>> file_prefix("data/single.gwf")
'single'

Source code in gwframe/pipeline.py

def file_prefix(path: str | PathLike[str]) -> str:
    """
    Return the ``OBSERVATORY-DESCRIPTION`` prefix of a frame file name.

    Files that don't follow the ``OBS-DESC-GPS-DURATION.gwf`` convention
    use their whole stem as the prefix.

    Examples
    --------
    >>> file_prefix("L-L1_HOFT_C00-1238166018-4096.gwf")
    'L-L1_HOFT_C00'
    >>> file_prefix("data/single.gwf")
    'single'
    """
    name = Path(path).name
    match = _FRAME_FILE_RE.match(name)
    if match:
        return match.group("prefix")
    return Path(name).stem

sort_files_by_gps

sort_files_by_gps(files: Iterable[str | PathLike[str]]) -> list[Path]

Sort frame files by the GPS start of their first frame.

Only file headers are read. Files without frames sort last, in name order.

Source code in gwframe/pipeline.py

def sort_files_by_gps(files: Iterable[str | PathLike[str]]) -> list[Path]:
    """
    Sort frame files by the GPS start of their first frame.

    Only file headers are read. Files without frames sort last, in name order.
    """

    def key(path: Path) -> tuple[float, str]:
        info = get_info(path)
        start = info.frames[0].start if info.frames else math.inf
        return (start, path.name)

    return sorted((Path(f) for f in files), key=key)