gwframe

Frame ¶

Frame(start: float, duration: float, name: str = '', run: int = 0, frame_number: int = 0, frame_spec: int | None = None)

Bases: MutableMapping

High-level interface for creating and manipulating GWF frames.

This class provides a Pythonic interface to the underlying frameCPP FrameH class, with simplified methods for adding data and metadata.

Parameters:

Name	Type	Description	Default
`start`	`float`	GPS start time of the frame	required
`duration`	`float`	Duration of the frame in seconds	required
`name`	`str`	Frame name (e.g., 'L1' for LIGO Livingston)	`''`
`run`	`int`	Run number (default: 0, negative for simulated data)	`0`

Notes

Detector information is automatically added to the frame based on channel names. When you add a channel with a name like 'L1:TEST', the detector information for L1 will be automatically included.

Examples:

>>> frame = gwframe.Frame(start=1234567890.0, duration=1.0, name='L1', run=1)
>>> frame.add_channel('L1:TEST', data=np.random.randn(16384),
...                   dt=1.0/16384, unit='counts')
>>> frame.write('output.gwf')

Source code in gwframe/write.py

def __init__(
    self,
    start: float,
    duration: float,
    name: str = "",
    run: int = 0,
    frame_number: int = 0,
    frame_spec: int | None = None,
):
    # Convert GPS time
    self._gps_time = _core.gpstime_from_float(start)

    # Get leap seconds from GPS time (TAI-UTC offset)
    leap_seconds = self._gps_time.get_leap_seconds()

    # Use full constructor with frame_number and leap_seconds
    self._frame = _core.FrameH(
        name, run, frame_number, self._gps_time, leap_seconds, duration
    )

    # Store for convenience
    self._start = start
    self.duration = duration
    self.name = name
    self.run = run
    self.frame_number = frame_number
    self._frame_spec = frame_spec

    # Keep references to vects to prevent premature garbage collection
    # (C++ frame holds raw pointers, so Python must keep objects alive)
    self._vects: list[_core.FrVect] = []
    self._frdatas: list[_core.FrProcData | _core.FrAdcData | _core.FrSimData] = []

    # Track which detectors we've added to avoid duplicates
    self._detectors_added: set[str] = set()

    # Store channels for dict-like access
    self._channels: dict[str, TimeSeries] = {}

    # Track if channels were modified via dict interface
    self._channels_modified = False

start `property` `writable` ¶

start: float

GPS start time of the frame.

delitem ¶

__delitem__(key: str) -> None

Delete channel by name. Frame will be rebuilt on write.

Source code in gwframe/write.py

def __delitem__(self, key: str) -> None:
    """Delete channel by name. Frame will be rebuilt on write."""
    del self._channels[key]
    self._channels_modified = True

getitem ¶

__getitem__(key: str) -> TimeSeries

Get channel by name.

Source code in gwframe/write.py

def __getitem__(self, key: str) -> TimeSeries:
    """Get channel by name."""
    return self._channels[key]

iter ¶

__iter__()

Iterate over channel names.

Source code in gwframe/write.py

def __iter__(self):
    """Iterate over channel names."""
    return iter(self._channels)

len ¶

__len__() -> int

Return number of channels.

Source code in gwframe/write.py

def __len__(self) -> int:
    """Return number of channels."""
    return len(self._channels)

setitem ¶

__setitem__(key: str, value: TimeSeries) -> None

Set/update channel with TimeSeries.

Source code in gwframe/write.py

def __setitem__(self, key: str, value: TimeSeries) -> None:
    """Set/update channel with TimeSeries."""
    self._channels[key] = value
    self._channels_modified = True

add_channel ¶

add_channel(channel: str, data: NDArray | MaskedArray, sample_rate: float, unit: str = '', comment: str = '', channel_type: str = 'proc', on_mask_loss: str | OnMaskLoss = WARN)

Add a data channel to this frame.

Parameters:

Name	Type	Description	Default
`channel`	`str`	Channel name (e.g., 'L1:TEST-CHANNEL')	required
`data`	`ndarray`	1D NumPy array containing the channel data	required
`sample_rate`	`float`	Sample rate in Hz (samples per second)	required
`unit`	`str`	Physical unit of the data (e.g., 'strain', 'counts')	`''`
`comment`	`str`	Comment or description for this channel	`''`
`channel_type`	`str`	Type of channel: 'proc' (processed, default) or 'sim' (simulated)	`'proc'`

Examples:

>>> frame.add_channel('L1:TEST', data=np.random.randn(16384),
...                   sample_rate=16384, unit='counts')

Notes

The data type (float64, float32, int32, etc.) is automatically determined from the NumPy array dtype.

Source code in gwframe/write.py

def add_channel(
    self,
    channel: str,
    data: npt.NDArray | np.ma.MaskedArray,
    sample_rate: float,
    unit: str = "",
    comment: str = "",
    channel_type: str = "proc",
    on_mask_loss: str | OnMaskLoss = OnMaskLoss.WARN,
):
    """
    Add a data channel to this frame.

    Parameters
    ----------
    channel : str
        Channel name (e.g., 'L1:TEST-CHANNEL')
    data : np.ndarray
        1D NumPy array containing the channel data
    sample_rate : float
        Sample rate in Hz (samples per second)
    unit : str, optional
        Physical unit of the data (e.g., 'strain', 'counts')
    comment : str, optional
        Comment or description for this channel
    channel_type : str, optional
        Type of channel: 'proc' (processed, default) or 'sim' (simulated)

    Examples
    --------
    >>> frame.add_channel('L1:TEST', data=np.random.randn(16384),
    ...                   sample_rate=16384, unit='counts')

    Notes
    -----
    The data type (float64, float32, int32, etc.) is automatically
    determined from the NumPy array dtype.
    """
    # Ensure data is a 1D numpy array
    if data.ndim != 1:
        msg = f"Data must be 1D array, got shape {data.shape}"
        raise ValueError(msg)

    # Handle masked arrays: extract mask and strip to plain ndarray
    import warnings

    mask = None
    if isinstance(data, np.ma.MaskedArray):
        mask = np.ma.getmaskarray(data)
        data = np.asarray(data)

    on_mask_loss = OnMaskLoss(on_mask_loss)

    # Extract detector prefix from channel name (e.g., 'L1:TEST' -> 'L1')
    # and add detector information if it's a known detector
    if ":" in channel:
        prefix = channel.split(":", 1)[0]
        if prefix not in self._detectors_added:
            try:
                # Check if this is a valid detector
                detector_loc = DetectorLocation[prefix]
                # Get detector info for this location and GPS time
                detector = _core.get_detector(detector_loc, self._gps_time)
                # Append to frame based on channel type
                if channel_type == "sim":
                    self._frame.append_fr_detector_sim(detector)
                else:  # 'proc' or 'adc'
                    self._frame.append_fr_detector_proc(detector)
                # Mark this detector as added
                self._detectors_added.add(prefix)
            except KeyError:
                # Not a known detector prefix, that's okay
                pass

    n_samples = len(data)

    # Convert sample_rate to dt (sample spacing)
    dt = 1.0 / sample_rate

    if data.dtype not in _DTYPE_TO_FRVECT:
        msg = (
            f"Unsupported data type: {data.dtype}. "
            f"Supported types: {list(_DTYPE_TO_FRVECT.keys())}"
        )
        raise ValueError(msg)

    frvect_type = _DTYPE_TO_FRVECT[data.dtype]

    # Create dimension
    dim = _core.Dimension(n_samples, dt, "s", 0.0)

    # Create FrVect and populate with data
    vect = _core.FrVect(channel, frvect_type, 1, dim, unit)
    # Direct C++ memcpy ~50% faster than get_data_array()[:] = data
    vect.set_data(data)

    # Create appropriate Fr*Data container and add to frame
    if channel_type == "proc":
        # Calculate Nyquist frequency (frange) from sample rate
        frange = sample_rate / 2.0  # Nyquist frequency

        # FrProcData: name, comment, type, subtype, time_offset, trange,
        # fshift, phase, frange, bandwidth
        frdata = _core.FrProcData(
            channel, comment, 1, 0, 0.0, self.duration, 0.0, 0.0, frange, 0.0
        )
        frdata.append_data(vect)
        self._frame.append_fr_proc_data(frdata)
    elif channel_type == "adc":
        # Determine nbits from dtype
        nbits = data.dtype.itemsize * 8

        # FrAdcData: name, channelgroup, channelid, nbits, sample_rate
        frdata = _core.FrAdcData(channel, 0, 0, nbits, sample_rate)
        frdata.append_data(vect)

        # Set channel-level dataValid before appending to frame
        # (append copies the FrAdcData, so modifications after append won't persist)
        if mask is not None and np.any(mask):
            frdata.set_data_valid(1)

        self._frame.append_fr_adc_data(frdata)
    elif channel_type == "sim":
        # FrSimData: name, comment, sample_rate, time_offset, fshift, phase
        frdata = _core.FrSimData(channel, comment, sample_rate, 0.0, 0.0, 0.0)
        frdata.append_data(vect)
        self._frame.append_fr_sim_data(frdata)
    else:
        msg = (
            f"Unsupported channel_type: {channel_type}. "
            f"Supported types: 'proc', 'adc', 'sim'"
        )
        raise ValueError(msg)

    # Handle mask loss warnings/errors
    if mask is not None and np.any(mask):
        mask_loss_msg = None

        if channel_type == "adc":
            # dataValid already set above before append
            if not np.all(mask):
                mask_loss_msg = (
                    f"Masked array for ADC channel '{channel}' has a "
                    f"per-sample mask, but this frame format only supports "
                    f"per-channel masking. The entire channel will be "
                    f"flagged as invalid."
                )
        else:
            mask_loss_msg = (
                f"Masked array passed for {channel_type} channel "
                f"'{channel}', but masking is not supported for "
                f"{channel_type} channels on this frame format version. "
                f"The mask will be discarded."
            )

        if mask_loss_msg is not None:
            match on_mask_loss:
                case OnMaskLoss.RAISE:
                    raise ValueError(mask_loss_msg)
                case OnMaskLoss.WARN:
                    warnings.warn(mask_loss_msg, UserWarning, stacklevel=2)
                case OnMaskLoss.IGNORE:
                    pass

    # Keep references alive to prevent garbage collection
    # (C++ uses raw pointers with empty deleters, so Python must keep objects alive)
    self._vects.append(vect)
    self._frdatas.append(frdata)

    # Also store in channels dict for dict-like access
    self._channels[channel] = TimeSeries(
        array=data,
        name=channel,
        dtype=data.dtype,
        start=self.start,
        dt=dt,
        duration=len(data) * dt,
        sample_rate=sample_rate,
        unit=unit,
        type=channel_type,
        mask=mask,
    )

add_history ¶

add_history(name: str, comment: str, time: int | None = None)

Add a history/metadata entry to the frame.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name/key for this metadata entry	required
`comment`	`str`	The metadata value/comment	required
`time`	`int`	GPS time for this entry (default: frame start time)	`None`

Source code in gwframe/write.py

def add_history(self, name: str, comment: str, time: int | None = None):
    """
    Add a history/metadata entry to the frame.

    Parameters
    ----------
    name : str
        Name/key for this metadata entry
    comment : str
        The metadata value/comment
    time : int, optional
        GPS time for this entry (default: frame start time)
    """
    if time is None:
        time = int(self.start)
    history = _core.FrHistory(name, time, comment)
    self._frame.append_frhistory(history)

write ¶

write(filename: str | PathLike[str], compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6)

Write this frame to a GWF file.

Parameters:

Name	Type	Description	Default
`filename`	`str or path - like`	Output file path	required
`compression`	`int`	Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP) Use Compression.RAW for no compression	`ZERO_SUPPRESS_OTHERWISE_GZIP`
`compression_level`	`int`	Compression level 0-9 (default: 6, higher = more compression)	`6`

Examples:

>>> frame.write('output.gwf')
>>> frame.write('output_raw.gwf', compression=gwframe.Compression.RAW)

Source code in gwframe/write.py

def write(
    self,
    filename: str | PathLike[str],
    compression: int = Compression.ZERO_SUPPRESS_OTHERWISE_GZIP,
    compression_level: int = 6,
):
    """
    Write this frame to a GWF file.

    Parameters
    ----------
    filename : str or path-like
        Output file path
    compression : int, optional
        Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)
        Use Compression.RAW for no compression
    compression_level : int, optional
        Compression level 0-9 (default: 6, higher = more compression)

    Examples
    --------
    >>> frame.write('output.gwf')
    >>> frame.write('output_raw.gwf', compression=gwframe.Compression.RAW)
    """
    # Rebuild frame if channels were modified via dict interface
    if self._channels_modified:
        self._rebuild_frame()

    # Write to a temp file, then atomically move to the final path
    dest = fspath(filename)
    dir_name = os.path.dirname(dest) or "."
    fd, tmp_path = tempfile.mkstemp(suffix=".gwf.tmp", dir=dir_name)
    os.close(fd)
    try:
        self._frame.write(
            _core.OFrameFStream(tmp_path), compression, compression_level
        )
        shutil.move(tmp_path, dest)
    except BaseException:
        with contextlib.suppress(OSError):
            os.unlink(tmp_path)
        raise

write_bytes ¶

write_bytes(compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6) -> bytes

Write this frame to bytes (in-memory GWF format).

Parameters:

Name	Type	Description	Default
`compression`	`int`	Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)	`ZERO_SUPPRESS_OTHERWISE_GZIP`
`compression_level`	`int`	Compression level 0-9 (default: 6)	`6`

Returns:

Type	Description
`bytes`	GWF-formatted data as bytes

Examples:

>>> frame = gwframe.Frame(start=1234567890.0, duration=1.0, name='L1')
>>> frame.add_channel('L1:TEST', data, dt=1.0/16384)
>>> gwf_bytes = frame.write_bytes()
>>> # Verify round-trip
>>> read_data = gwframe.read_bytes(gwf_bytes, 'L1:TEST')

Source code in gwframe/write.py

def write_bytes(
    self,
    compression: int = Compression.ZERO_SUPPRESS_OTHERWISE_GZIP,
    compression_level: int = 6,
) -> bytes:
    """
    Write this frame to bytes (in-memory GWF format).

    Parameters
    ----------
    compression : int, optional
        Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)
    compression_level : int, optional
        Compression level 0-9 (default: 6)

    Returns
    -------
    bytes
        GWF-formatted data as bytes

    Examples
    --------
    >>> frame = gwframe.Frame(start=1234567890.0, duration=1.0, name='L1')
    >>> frame.add_channel('L1:TEST', data, dt=1.0/16384)
    >>> gwf_bytes = frame.write_bytes()
    >>> # Verify round-trip
    >>> read_data = gwframe.read_bytes(gwf_bytes, 'L1:TEST')
    """
    # Rebuild frame if channels were modified via dict interface
    if self._channels_modified:
        self._rebuild_frame()

    # Create memory buffer for output
    buffer = _core.MemoryBuffer(_core.IOS_OUT)

    # Write frame in a scope to ensure stream is destroyed (flushed) before reading
    # The stream destructor writes the TOC which is critical for reading
    stream = _core.OFrameMemStream(buffer)
    self._frame.write(stream, compression, compression_level)
    del stream  # Explicitly destroy stream to flush TOC

    # Extract bytes from buffer
    return buffer.get_bytes()

write_to_stream ¶

write_to_stream(stream, compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6)

Write this frame to an output stream.

This is used internally by FrameWriter for writing multiple frames.

Parameters:

Name	Type	Description	Default
`stream`	`OFrameFStream`	Output stream to write to	required
`compression`	`int`	Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)	`ZERO_SUPPRESS_OTHERWISE_GZIP`
`compression_level`	`int`	Compression level 0-9 (default: 6)	`6`

Source code in gwframe/write.py

def write_to_stream(
    self,
    stream,
    compression: int = Compression.ZERO_SUPPRESS_OTHERWISE_GZIP,
    compression_level: int = 6,
):
    """
    Write this frame to an output stream.

    This is used internally by FrameWriter for writing multiple frames.

    Parameters
    ----------
    stream : OFrameFStream
        Output stream to write to
    compression : int, optional
        Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)
    compression_level : int, optional
        Compression level 0-9 (default: 6)
    """
    # Rebuild frame if channels were modified via dict interface
    if self._channels_modified:
        self._rebuild_frame()

    self._frame.write(stream, compression, compression_level)

FrameWriter ¶

FrameWriter(destination: str | PathLike[str] | BytesIO, compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6, frame_number: int = 0, frame_spec: int | None = None)

Context manager for writing multiple frames to a GWF file or BytesIO buffer.

This is the recommended way to write multiple frames, as it keeps the output stream open and efficiently writes frames sequentially.

Parameters:

Name	Type	Description	Default
`destination`	`str, path-like, or BytesIO`	Output destination - either a file path or BytesIO object	required
`compression`	`int`	Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)	`ZERO_SUPPRESS_OTHERWISE_GZIP`
`compression_level`	`int`	Compression level 0-9 (default: 6)	`6`

Examples:

>>> # Write multiple 1-second frames to file
>>> with gwframe.FrameWriter('output.gwf') as writer:
...     for i in range(10):
...         start = 1234567890.0 + i
...         data = np.random.randn(16384)
...         writer.write(data, start=start, sample_rate=16384, name='L1:TEST')

>>> # Write to BytesIO
>>> from io import BytesIO
>>> buffer = BytesIO()
>>> with gwframe.FrameWriter(buffer) as writer:
...     for i in range(10):
...         data = np.random.randn(16384)
...         writer.write(data, start=1234567890.0 + i,
...                      sample_rate=16384, name='L1:TEST')
>>> gwf_bytes = buffer.getvalue()

Source code in gwframe/write.py

def __init__(
    self,
    destination: str | PathLike[str] | BytesIO,
    compression: int = Compression.ZERO_SUPPRESS_OTHERWISE_GZIP,
    compression_level: int = 6,
    frame_number: int = 0,
    frame_spec: int | None = None,
):
    self.compression = compression
    self.compression_level = compression_level
    self._frame_number = frame_number
    self._frame_spec = frame_spec
    self._stream = None
    self._memory_buffer = None
    self._bytesio_dest = None
    self._tmp_path: str | None = None

    # Determine if destination is file or BytesIO
    if isinstance(destination, BytesIO):
        self._bytesio_dest = destination
        self.filename = None
    else:
        self.filename = fspath(destination)
        self._bytesio_dest = None

close ¶

close()

Finalize and close the writer.

For file destinations, this flushes the stream and atomically moves the temporary file to the final path. For BytesIO destinations, this extracts the bytes and writes them to the buffer.

This method is idempotent — calling it on an already-closed writer is a no-op.

Source code in gwframe/write.py

def close(self):
    """
    Finalize and close the writer.

    For file destinations, this flushes the stream and atomically moves the
    temporary file to the final path. For BytesIO destinations, this
    extracts the bytes and writes them to the buffer.

    This method is idempotent — calling it on an already-closed writer is a
    no-op.
    """
    if self._stream is None:
        return

    if self._bytesio_dest is not None and self._memory_buffer is not None:
        self._stream = None
        gwf_bytes = self._memory_buffer.get_bytes()
        self._bytesio_dest.write(gwf_bytes)
    elif self._tmp_path is not None:
        assert self.filename is not None
        self._stream = None
        shutil.move(self._tmp_path, self.filename)

    self._stream = None
    self._memory_buffer = None
    self._tmp_path = None

open ¶

open()

Open the writer for writing frames.

This sets up the output stream. For file destinations, writes go to a temporary file that is atomically moved to the final path on :meth:close.

Can also be used via the context manager protocol (with statement), which calls :meth:open and :meth:close automatically.

Source code in gwframe/write.py

def open(self):
    """
    Open the writer for writing frames.

    This sets up the output stream. For file destinations, writes go to a
    temporary file that is atomically moved to the final path on
    :meth:`close`.

    Can also be used via the context manager protocol (``with`` statement),
    which calls :meth:`open` and :meth:`close` automatically.
    """
    if self._stream is not None:
        msg = "FrameWriter is already open"
        raise RuntimeError(msg)

    if self._bytesio_dest is not None:
        self._memory_buffer = _core.MemoryBuffer(_core.IOS_OUT)
        self._stream = _core.OFrameMemStream(self._memory_buffer)
    else:
        assert self.filename is not None
        dir_name = os.path.dirname(self.filename) or "."
        fd, self._tmp_path = tempfile.mkstemp(suffix=".gwf.tmp", dir=dir_name)
        os.close(fd)
        self._stream = _core.OFrameFStream(self._tmp_path)

    return self

write ¶

write(channels: dict[str, NDArray] | NDArray, start: float, sample_rate: float | dict[str, float], *, name: str = '', run: int = 0, unit: str | dict[str, str] = '', channel_type: str = 'proc', on_mask_loss: str | OnMaskLoss = WARN)

Convenience method to write data directly without creating Frame object.

This creates a Frame internally and writes it immediately.

Parameters:

Name	Type	Description	Default
`channels`	`dict or ndarray`	Channel data. Either: - dict mapping channel names to 1D NumPy arrays - Single 1D NumPy array (requires channel name in name parameter)	required
`start`	`float`	GPS start time of the frame	required
`sample_rate`	`float or dict`	Sample rate in Hz. Either: - Single float value used for all channels - dict mapping channel names to sample rates	required
`name`	`str`	Frame name (e.g., 'L1') or single channel name if channels is an array	`''`
`run`	`int`	Run number (default: 0, negative for simulated data)	`0`
`unit`	`str or dict`	Physical unit. Either: - Single string used for all channels (default: '') - dict mapping channel names to units	`''`
`channel_type`	`str`	Type of channels: 'proc' (processed, default) or 'sim' (simulated)	`'proc'`

Examples:

>>> with gwframe.FrameWriter('output.gwf') as writer:
...     for i in range(10):
...         data = np.random.randn(16384)
...         writer.write(
...             data, start=1234567890.0 + i, sample_rate=16384, name='L1:TEST'
...         )

Source code in gwframe/write.py

def write(
    self,
    channels: dict[str, npt.NDArray] | npt.NDArray,
    start: float,
    sample_rate: float | dict[str, float],
    *,
    name: str = "",
    run: int = 0,
    unit: str | dict[str, str] = "",
    channel_type: str = "proc",
    on_mask_loss: str | OnMaskLoss = OnMaskLoss.WARN,
):
    """
    Convenience method to write data directly without creating Frame object.

    This creates a Frame internally and writes it immediately.

    Parameters
    ----------
    channels : dict or np.ndarray
        Channel data. Either:
        - dict mapping channel names to 1D NumPy arrays
        - Single 1D NumPy array (requires channel name in name parameter)
    start : float
        GPS start time of the frame
    sample_rate : float or dict
        Sample rate in Hz. Either:
        - Single float value used for all channels
        - dict mapping channel names to sample rates
    name : str, optional
        Frame name (e.g., 'L1') or single channel name if channels is an array
    run : int, optional
        Run number (default: 0, negative for simulated data)
    unit : str or dict, optional
        Physical unit. Either:
        - Single string used for all channels (default: '')
        - dict mapping channel names to units
    channel_type : str, optional
        Type of channels: 'proc' (processed, default) or 'sim' (simulated)

    Examples
    --------
    >>> with gwframe.FrameWriter('output.gwf') as writer:
    ...     for i in range(10):
    ...         data = np.random.randn(16384)
    ...         writer.write(
    ...             data, start=1234567890.0 + i, sample_rate=16384, name='L1:TEST'
    ...         )
    """
    if self._stream is None:
        msg = "FrameWriter not opened (call open() or use 'with' statement)"
        raise RuntimeError(msg)

    # Handle single array case - convert to dict
    if isinstance(channels, np.ndarray):
        if not name:
            msg = "name parameter required when channels is a single array"
            raise ValueError(msg)
        channel_name = name
        channels = {channel_name: channels}
        frame_name = channel_name.split(":")[0] if ":" in channel_name else ""
    else:
        frame_name = name

    # Determine frame duration from first channel
    first_channel = next(iter(channels.keys()))
    first_data = channels[first_channel]
    first_rate = (
        sample_rate
        if isinstance(sample_rate, (int, float))
        else sample_rate[first_channel]
    )
    duration = len(first_data) / first_rate

    # Create frame with auto-incremented frame number
    frame = Frame(
        start=start,
        duration=duration,
        name=frame_name,
        run=run,
        frame_number=self._frame_number,
        frame_spec=self._frame_spec,
    )

    # Add all channels to the frame
    _populate_frame_with_channels(
        frame, channels, sample_rate, unit, channel_type, on_mask_loss
    )

    # Write the frame (this will auto-increment _frame_number)
    self.write_frame(frame)

write_frame ¶

write_frame(frame: Frame)

Write a Frame object to the file.

Parameters:

Name	Type	Description	Default
`frame`	`Frame`	The frame to write	required

Examples:

>>> with gwframe.FrameWriter('output.gwf') as writer:
...     frame = gwframe.Frame(start=1234567890.0, duration=1.0, name='L1')
...     frame.add_channel('L1:TEST', data, dt=1.0/16384)
...     writer.write_frame(frame)

Notes

If the frame was created with frame_number=0 (default), the writer will use its tracked frame_number. Otherwise, the frame's frame_number is used. Frame numbers auto-increment with each write.

Source code in gwframe/write.py

def write_frame(self, frame: Frame):
    """
    Write a Frame object to the file.

    Parameters
    ----------
    frame : Frame
        The frame to write

    Examples
    --------
    >>> with gwframe.FrameWriter('output.gwf') as writer:
    ...     frame = gwframe.Frame(start=1234567890.0, duration=1.0, name='L1')
    ...     frame.add_channel('L1:TEST', data, dt=1.0/16384)
    ...     writer.write_frame(frame)

    Notes
    -----
    If the frame was created with frame_number=0 (default), the writer
    will use its tracked frame_number. Otherwise, the frame's frame_number
    is used. Frame numbers auto-increment with each write.
    """
    if self._stream is None:
        msg = "FrameWriter not opened (call open() or use 'with' statement)"
        raise RuntimeError(msg)

    frame.write_to_stream(self._stream, self.compression, self.compression_level)

    # Auto-increment for next frame
    self._frame_number += 1

TimeSeries `dataclass` ¶

TimeSeries(array: NDArray[floating], name: str, dtype: dtype, start: float, dt: float, duration: float, sample_rate: float, unit: str, type: str, mask: NDArray[bool_] | None = None)

Time series data from a GWF channel.

This dataclass holds the array data and metadata for a channel read from a GWF file.

Attributes:

Name	Type	Description
`array`	`ndarray`	NumPy array containing the time series data
`name`	`str`	Channel name (e.g., 'H1:LOSC-STRAIN')
`dtype`	`dtype`	NumPy dtype of the samples. Always mirrors `array.dtype`.
`start`	`float`	Start time in GPS seconds
`dt`	`float`	Sample spacing in seconds
`duration`	`float`	Total duration in seconds
`sample_rate`	`float`	Sampling rate in Hz (1/dt)
`unit`	`str`	Physical unit of the data (e.g., 'strain')
`type`	`str`	Channel type: 'proc' (processed), 'adc' (raw ADC), or 'sim' (simulated)
`mask`	`ndarray or None`	Boolean mask where `True` indicates invalid samples. `None` when all samples are valid. Only populated when reading with `allow_invalid=True` and the file contains dataValid flags.

Examples:

>>> data = gwframe.read('data.gwf', 'H1:LOSC-STRAIN')
>>> print(f"Channel: {data.name}")
>>> print(f"Duration: {data.duration} s at {data.sample_rate} Hz")
>>> print(f"Data shape: {data.array.shape}")
>>> # Check for invalid samples
>>> if data.mask is not None:
...     print(f"Invalid samples: {data.mask.sum()}")

to_masked ¶

to_masked() -> NDArray[floating] | MaskedArray

Return data as a masked array if a mask is present, otherwise plain array.

This is useful when passing data to APIs that understand masked arrays, or when you want numpy operations to automatically skip invalid samples.

Returns:

Type	Description
`MaskedArray or ndarray`	Masked array if `mask` is set, plain array otherwise.

Source code in gwframe/types.py

def to_masked(self) -> npt.NDArray[np.floating] | np.ma.MaskedArray:
    """Return data as a masked array if a mask is present, otherwise plain array.

    This is useful when passing data to APIs that understand masked arrays,
    or when you want numpy operations to automatically skip invalid samples.

    Returns
    -------
    np.ma.MaskedArray or np.ndarray
        Masked array if ``mask`` is set, plain array otherwise.
    """
    if self.mask is not None:
        return np.ma.MaskedArray(self.array, mask=self.mask)
    return self.array

read_bytes ¶

read_bytes(data: bytes, channels: str, frame_index: int = 0, *, validate_checksum: bool = False, start: float | None = None, end: float | None = None, allow_invalid: bool = False) -> TimeSeries

read_bytes(data: bytes, channels: None, frame_index: int = 0, *, validate_checksum: bool = False, start: float | None = None, end: float | None = None, allow_invalid: bool = False) -> dict[str, TimeSeries]

read_bytes(data: bytes, channels: list[str], frame_index: int = 0, *, validate_checksum: bool = False, start: float | None = None, end: float | None = None, allow_invalid: bool = False) -> dict[str, TimeSeries]

read_bytes(data: bytes, channels: str | None | list[str] = None, frame_index: int = 0, *, validate_checksum: bool = False, start: float | None = None, end: float | None = None, allow_invalid: bool = False) -> TimeSeries | dict[str, TimeSeries]

Read channel data from GWF data in memory (bytes).

This allows reading GWF data without writing to disk first, which is useful when working with data from network streams, compressed archives, or in-memory buffers.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	Raw GWF file data as bytes	required
`channels`	`str, None, or list[str]`	Channel(s) to read: - str: Read single channel (e.g., 'L1:GWOSC-16KHZ_R1_STRAIN') - None: Read all channels from the frame (default) - list[str]: Read specific list of channels	`None`
`frame_index`	`int`	Index of the frame to read from (default: 0)	`0`
`validate_checksum`	`bool`	Validate frame file checksums before reading (default: False). When enabled, performs file-level checksum validation which requires reading the entire frame file. Disabled by default for performance.	`False`

Returns:

Name	Type	Description
`data`	`TimeSeries or dict[str, TimeSeries]`	If channels is a str: returns TimeSeries for that channel If channels is None or list[str]: returns dict mapping channel names to TimeSeries

Examples:

>>> with open('data.gwf', 'rb') as f:
...     gwf_bytes = f.read()
>>> data = gwframe.read_bytes(gwf_bytes, 'L1:GWOSC-16KHZ_R1_STRAIN')
>>> print(f"Read {len(data.array)} samples at {data.sample_rate} Hz")

>>> # Read all channels
>>> all_data = gwframe.read_bytes(gwf_bytes, channels=None)
>>> print(f"Found {len(all_data)} channels")

>>> import io
>>> from io import BytesIO
>>> data = gwframe.read_bytes(BytesIO(gwf_bytes).read(), 'L1:STRAIN')

Notes

This function uses frameCPP's MemoryBuffer internally to read from memory without writing to disk.

Source code in gwframe/read.py

def read_bytes(
    data: bytes,
    channels: str | None | list[str] = None,
    frame_index: int = 0,
    *,
    validate_checksum: bool = False,
    start: float | None = None,
    end: float | None = None,
    allow_invalid: bool = False,
) -> TimeSeries | dict[str, TimeSeries]:
    """
    Read channel data from GWF data in memory (bytes).

    This allows reading GWF data without writing to disk first,
    which is useful when working with data from network streams,
    compressed archives, or in-memory buffers.

    Parameters
    ----------
    data : bytes
        Raw GWF file data as bytes
    channels : str, None, or list[str], optional
        Channel(s) to read:
        - str: Read single channel (e.g., 'L1:GWOSC-16KHZ_R1_STRAIN')
        - None: Read all channels from the frame (default)
        - list[str]: Read specific list of channels
    frame_index : int, optional
        Index of the frame to read from (default: 0)
    validate_checksum : bool, optional
        Validate frame file checksums before reading (default: False).
        When enabled, performs file-level checksum validation which requires
        reading the entire frame file. Disabled by default for performance.

    Returns
    -------
    data : TimeSeries or dict[str, TimeSeries]
        - If channels is a str: returns TimeSeries for that channel
        - If channels is None or list[str]: returns dict mapping channel names
          to TimeSeries

    Examples
    --------
    >>> with open('data.gwf', 'rb') as f:
    ...     gwf_bytes = f.read()
    >>> data = gwframe.read_bytes(gwf_bytes, 'L1:GWOSC-16KHZ_R1_STRAIN')
    >>> print(f"Read {len(data.array)} samples at {data.sample_rate} Hz")

    >>> # Read all channels
    >>> all_data = gwframe.read_bytes(gwf_bytes, channels=None)
    >>> print(f"Found {len(all_data)} channels")

    >>> import io
    >>> from io import BytesIO
    >>> data = gwframe.read_bytes(BytesIO(gwf_bytes).read(), 'L1:STRAIN')

    Notes
    -----
    This function uses frameCPP's MemoryBuffer internally to read
    from memory without writing to disk.
    """
    # Verify input is bytes
    if not isinstance(data, bytes):
        msg = f"data must be bytes, got {type(data)}"
        raise TypeError(msg)

    # Validate time-based slicing parameters
    if (start is None) != (end is None):
        msg = "start and end must be specified together"
        raise ValueError(msg)

    if start is not None and frame_index != 0:
        msg = "start/end parameters are mutually exclusive with frame_index"
        raise ValueError(msg)

    # Bounds-check frame_index before any channel enumeration or read, so
    # callers see a FrameIndexError with proper context instead of a raw
    # C-extension error or a misleading ChannelNotFoundError. The time-slice
    # path forces frame_index=0 and has its own InvalidTimeRangeError
    # handling, so skip the bounds check when start is set.
    if start is None:
        frame_times = _core.get_frame_times_from_bytes(data)
        num_frames = len(frame_times)
        if frame_index < 0 or frame_index >= num_frames:
            raise FrameIndexError(frame_index, num_frames)

    # Handle time-based slicing
    if start is not None:
        # For multi-channel, recursively read each channel with time slicing
        if channels is None or isinstance(channels, list):
            if channels is None:
                proc_ch, adc_ch, sim_ch = _core.enumerate_channels_from_bytes(data, 0)
                channels_to_read = list(proc_ch) + list(adc_ch) + list(sim_ch)
            else:
                channels_to_read = channels

            result: dict[str, TimeSeries] = {}
            for ch in channels_to_read:
                try:
                    ts = read_bytes(
                        data,
                        ch,
                        0,
                        validate_checksum=validate_checksum,
                        start=start,
                        end=end,
                        allow_invalid=allow_invalid,
                    )
                    assert isinstance(
                        ts, TimeSeries
                    )  # Single channel returns TimeSeries
                    result[ch] = ts
                except (ValueError, RuntimeError):
                    continue
            return result

        # Single channel case - must be a string
        if not isinstance(channels, str):
            msg = f"channels must be str, None, or list[str], got {type(channels)}"
            raise TypeError(msg)

        # Find frames overlapping [start, end) using frame times
        frame_times = _core.get_frame_times_from_bytes(data)

        # Find which frames overlap with [start, end)
        # A frame overlaps if: frame_start < end AND frame_end > start
        frame_indices = [
            i
            for i, (frame_start, frame_duration) in enumerate(frame_times)
            if frame_start < end and (frame_start + frame_duration) > start
        ]

        if not frame_indices:
            # Get file time range for helpful error message
            if frame_times:
                file_start = frame_times[0][0]
                file_end = frame_times[-1][0] + frame_times[-1][1]
            else:
                file_start = file_end = 0.0
            assert start is not None and end is not None
            raise InvalidTimeRangeError(start, end, file_start, file_end)

        # Read and concatenate frames
        timeseries_list: list[TimeSeries] = []
        for frame_idx in frame_indices:
            ts = read_bytes(
                data,
                channels,
                frame_idx,
                validate_checksum=validate_checksum,
                allow_invalid=allow_invalid,
            )
            assert isinstance(ts, TimeSeries)  # Narrows type for mypy
            timeseries_list.append(ts)

        # Stitch and slice using shared helper
        assert start is not None and end is not None
        return _stitch_and_slice_timeseries(timeseries_list, start, end)

    # Handle multiple channel cases
    if channels is None or isinstance(channels, list):
        # Get list of channels to read
        if channels is None:
            # Read all channels - enumerate from frame
            proc_ch, adc_ch, sim_ch = _core.enumerate_channels_from_bytes(
                data, frame_index
            )
            channels_to_read = list(proc_ch) + list(adc_ch) + list(sim_ch)
        else:
            # channels is a list[str]
            channels_to_read = channels

        # Read each channel and return dict
        channel_dict: dict[str, TimeSeries] = {}
        for ch in channels_to_read:
            try:
                ts = read_bytes(
                    data,
                    ch,
                    frame_index,
                    validate_checksum=validate_checksum,
                    allow_invalid=allow_invalid,
                )
                assert isinstance(ts, TimeSeries)  # Single channel returns TimeSeries
                channel_dict[ch] = ts
            except (ValueError, RuntimeError):
                # Skip channels that fail to read
                continue

        return channel_dict

    # Single channel case (channels is a str)
    if not isinstance(channels, str):
        msg = f"channels must be str, None, or list[str], got {type(channels)}"
        raise TypeError(msg)

    # Validate checksums if requested
    if validate_checksum:
        _core.validate_frame_checksums(data)

    # Try each channel type until one works.
    # proc/sim return 6-tuple: (array, name, type, time_offset, dx, unit_y)
    # adc returns 7-tuple: (array, name, type, time_offset, dx, unit_y, data_valid)
    channel_data = None
    channel_type = None

    readers = [
        ("proc", _core.read_proc_from_bytes),
        ("sim", _core.read_sim_from_bytes),
        ("adc", _core.read_adc_from_bytes),
    ]

    for reader_type, reader_func in readers:
        try:
            channel_data = reader_func(data, frame_index, channels)
            channel_type = reader_type
            break
        except (RuntimeError, ValueError, IndexError):
            continue

    if channel_data is None:
        # Get available channels to provide helpful error
        proc_ch, adc_ch, sim_ch = _core.enumerate_channels_from_bytes(data, frame_index)
        available = list(proc_ch) + list(adc_ch) + list(sim_ch)
        raise ChannelNotFoundError(channels, available)

    if channel_type == "adc":
        array, name, _, time_offset, dt, unit_y, data_valid = channel_data
        if data_valid != 0:
            n_samples = len(array)
            if not allow_invalid:
                raise InvalidDataError(channels, n_samples, n_samples)
            adc_mask = np.ones(n_samples, dtype=bool)
            array = np.ma.MaskedArray(array, mask=adc_mask)
    else:
        array, name, _, time_offset, dt, unit_y = channel_data

    # Read frame header to get GPS start time
    time_s, time_ns = _core.read_frame_gps_time(data, frame_index)
    frame_t0 = float(time_s) + float(time_ns) * 1e-9

    # Calculate data start time (frame start + offset)
    data_t0 = frame_t0 + time_offset

    # Calculate duration and sample rate
    duration = dt * len(array) if dt > 0 else 0.0
    sample_rate = 1.0 / dt if dt > 0 else 0.0

    # Ensure channel_type is set (helps mypy type narrowing)
    assert channel_type is not None

    arr, mask = _split_masked(array)
    return TimeSeries(
        array=arr,
        name=name,
        dtype=arr.dtype,
        start=data_t0,
        dt=dt,
        duration=duration,
        sample_rate=sample_rate,
        unit=unit_y,
        type=channel_type,
        mask=mask,
    )

read_frames ¶

read_frames(filename: str | PathLike[str], *, allow_invalid: bool = False) -> Generator[Frame, None, None]

Read frames from a GWF file, preserving complete metadata.

Yields Frame objects that can be written directly to disk with identical metadata (frame name, run number, frame number, etc.).

Parameters:

Name	Type	Description	Default
`filename`	`str or path - like`	Path to the GWF file	required

Yields:

Name	Type	Description
`frame`	`Frame`	Frame object containing all channel data with correct sample rates, units, types, and original frame metadata

Examples:

>>> # Iterate over frames
>>> for frame in gwframe.read_frames('data.gwf'):
...     print(f"Frame {frame.name} at GPS {frame.start}")

>>> # Process and write frames
>>> with gwframe.FrameWriter('output.gwf') as writer:
...     for frame in gwframe.read_frames('input.gwf'):
...         writer.write_frame(frame)

>>> # Collect all frames into a list
>>> frames = list(gwframe.read_frames('data.gwf'))
>>> print(f"Read {len(frames)} frames")

write_bytes ¶

write_bytes(channels: dict[str, NDArray] | NDArray, start: float, sample_rate: float | dict[str, float], *, name: str = '', run: int = 0, unit: str | dict[str, str] = '', channel_type: str = 'proc', compression: int = ZERO_SUPPRESS_OTHERWISE_GZIP, compression_level: int = 6, on_mask_loss: str | OnMaskLoss = WARN, frame_spec: int | None = None) -> bytes

Write channel data to bytes (in-memory GWF format).

Parameters are identical to write() function.

Returns:

Type	Description
`bytes`	GWF-formatted data as bytes

Examples:

>>> data = np.sin(np.linspace(0, 2*np.pi, 16384))
>>> gwf_bytes = gwframe.write_bytes(
...     data, start=1234567890.0, sample_rate=16384, name='L1:TEST'
... )
>>> # Verify round-trip
>>> read_data = gwframe.read_bytes(gwf_bytes, 'L1:TEST')

gwframe

Frame ¶

start property writable ¶

__delitem__ ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__setitem__ ¶

add_channel ¶

add_history ¶

write ¶

write_bytes ¶

write_to_stream ¶

FrameWriter ¶

close ¶

open ¶

write ¶

write_frame ¶

TimeSeries dataclass ¶

to_masked ¶

read_bytes ¶

read_frames ¶

write_bytes ¶

start `property` `writable` ¶

delitem ¶

getitem ¶

iter ¶

len ¶

setitem ¶

TimeSeries `dataclass` ¶