Skip to content

test_pipeline

Tests for the gwframe.pipeline module.

dataset

dataset(tmp_path)

Four conventionally named files, two 8 s frames each, contiguous.

Source code in gwframe/tests/test_pipeline.py
@pytest.fixture
def dataset(tmp_path):
    """Four conventionally named files, two 8 s frames each, contiguous."""
    raw = tmp_path / "raw"
    raw.mkdir()
    t0 = 1400000000
    files = []
    for i in range(4):
        start = t0 + i * 16
        path = raw / f"L-L1_RAW-{start}-16.gwf"
        with FrameWriter(path) as writer:
            for j in range(2):
                frame_start = start + j * 8
                n = 8 * 16
                data = np.arange(n, dtype=np.float64) + (frame_start - t0) * 16
                data[3] = np.nan
                writer.write_frame(
                    make_frame(
                        frame_start,
                        8,
                        {
                            "L1:GDS-CALIB_STRAIN": data,
                            "L1:DEBUG": np.zeros(n, dtype=np.float32),
                            "L1:KEEP": np.ones(n, dtype=np.int32),
                        },
                        frame_number=i * 2 + j,
                    )
                )
        files.append(path)
    return files

make_frame

make_frame(start, duration, channels, sample_rate=16.0, frame_number=0, channel_type='proc')

Build a frame from a dict of channel name -> array.

Source code in gwframe/tests/test_pipeline.py
def make_frame(
    start, duration, channels, sample_rate=16.0, frame_number=0, channel_type="proc"
):
    """Build a frame from a dict of channel name -> array."""
    frame = Frame(
        start=start, duration=duration, name="L1", run=1, frame_number=frame_number
    )
    for name, data in channels.items():
        frame.add_channel(
            name, data, sample_rate, unit="counts", channel_type=channel_type
        )
    return frame