write

Write functions and Frame class for GWF files.

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] = {}

    # Per-channel aux mask encoding, so rebuilds preserve it
    self._aux_mask: dict[str, bool] = {}

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

frame_number `property` `writable` ¶

frame_number: int

Frame sequence number.

frame_spec `property` ¶

frame_spec: int | None

Frame specification version the frame is written with, if set.

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._aux_mask.pop(key, None)
    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, *, aux_mask: bool = False)

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'`
`on_mask_loss`	`str or OnMaskLoss`	What to do when a mask cannot be preserved in the frame format: 'warn' (default), 'raise', or 'ignore'. See `OnMaskLoss`. Channels with a rate divisible by 16 have their masks preserved in 1/16s blocks.	`WARN`
`aux_mask`	`bool`	ADC channels only. If True (default: False), a partially masked channel stores its mask in an auxiliary FrVect (the CDS `dataValid` convention: 16-bit integers, 0 = good) and the channel-level dataValid flag stays 0; `on_mask_loss` does not apply because the mask is preserved. The aux layout is derived from the sample rate: integer rates that are a multiple of 16 Hz use one value per 1/16-second block (a block is invalid if any sample in it is masked), any other rate uses one value per sample. A fully masked channel sets the channel-level flag and writes no aux vector, exactly as with `aux_mask=False`.	`False`

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,
    *,
    aux_mask: bool = False,
):
    """
    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)
    on_mask_loss : str or OnMaskLoss, optional
        What to do when a mask cannot be preserved in the frame format:
        'warn' (default), 'raise', or 'ignore'. See ``OnMaskLoss``.
        Channels with a rate divisible by 16 have their masks preserved
        in 1/16s blocks.
    aux_mask : bool, optional
        ADC channels only. If True (default: False), a partially masked
        channel stores its mask in an auxiliary FrVect (the CDS
        ``dataValid`` convention: 16-bit integers, 0 = good) and the
        channel-level dataValid flag stays 0; ``on_mask_loss`` does not
        apply because the mask is preserved. The aux layout is derived
        from the sample rate: integer rates that are a multiple of
        16 Hz use one value per 1/16-second block (a block is invalid
        if any sample in it is masked), any other rate uses one value
        per sample. A fully masked channel sets the channel-level flag
        and writes no aux vector, exactly as with ``aux_mask=False``.

    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)

    if aux_mask and channel_type != "adc":
        msg = (
            f"aux_mask=True is only supported for ADC channels, "
            f"not {channel_type!r}"
        )
        raise ValueError(msg)

    # 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

    # Set when the mask is preserved in an aux vector (no mask loss)
    aux_encoded = False

    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)

        # Attach mask information before appending to frame (append
        # copies the FrAdcData, so modifications after append won't
        # persist)
        if mask is not None and np.any(mask):
            if aux_mask and not np.all(mask):
                # Partial mask: preserve it in an aux vector following
                # the CDS convention (FrVect "dataValid", int16,
                # 0 = good), block-reduced conservatively: any masked
                # sample marks the whole block invalid.
                block = _aux_block_size(sample_rate, n_samples)
                aux_values = mask.reshape(-1, block).any(axis=1).astype(np.int16)
                aux_dim = _core.Dimension(len(aux_values), dt * block, "s", 0.0)
                aux_vect = _core.FrVect(
                    "dataValid", _core.FrVect.FR_VECT_2S, 1, aux_dim, ""
                )
                aux_vect.set_data(aux_values)
                frdata.append_aux(aux_vect)
                self._vects.append(aux_vect)
                aux_encoded = True
            else:
                # Fully masked (or aux encoding disabled): whole-channel
                # dataValid flag
                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 (not applicable when the mask was
    # preserved in an aux vector)
    if mask is not None and np.any(mask) and not aux_encoded:
        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)

    # Remember the aux encoding so frame rebuilds preserve it
    self._aux_mask[channel] = bool(aux_mask)

    # 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.
    # The stream is closed explicitly so a failure to write the table
    # of contents surfaces here rather than being lost on destruction.
    dest = fspath(filename)
    tmp_path = dest + ".tmp"
    try:
        with _translate_io_error(tmp_path):
            stream = _core.OFrameFStream(tmp_path)
            self._frame.write(stream, compression, compression_level)
            stream.close()
        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()

    with _translate_io_error(None):
        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.

Every frame in a GWF file must contain the same set of channels — the file's table of contents stores one position per frame for every channel, so differing channel sets are unrepresentable. The writer enforces this: writing a frame whose channel set differs from the frames already written raises :class:ValueError.

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
    self._channel_names: frozenset[str] | None = None
    self._channel_names_lock = threading.Lock()

    # 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 writes the table of contents, closes 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.

Raises:

Type	Description
`OSError`	The table of contents could not be written (for example the disk is full). The temporary file is removed and the writer is closed; no output file is produced.

Source code in gwframe/write.py

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

    For file destinations, this writes the table of contents, closes
    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.

    Raises
    ------
    OSError
        The table of contents could not be written (for example the
        disk is full). The temporary file is removed and the writer
        is closed; no output file is produced.
    """
    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
        stream, self._stream = self._stream, None
        try:
            with _translate_io_error(self._tmp_path):
                stream.close()
        except BaseException:
            self._remove_tmp()
            raise
        shutil.move(self._tmp_path, self.filename)

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

discard ¶

discard()

Abandon the file being written without producing any output.

Use this instead of :meth:close after a write fails: close would finalize the truncated file and move it into place. The stream is closed (a failure to write the table of contents is expected here and ignored), the temporary file is removed, and the writer can be opened again. Idempotent.

Source code in gwframe/write.py

def discard(self):
    """Abandon the file being written without producing any output.

    Use this instead of :meth:`close` after a write fails: ``close``
    would finalize the truncated file and move it into place. The
    stream is closed (a failure to write the table of contents is
    expected here and ignored), the temporary file is removed, and
    the writer can be opened again. Idempotent.
    """
    if self._stream is None:
        return

    stream, self._stream = self._stream, None
    self._memory_buffer = None

    if self._tmp_path is not None:
        with contextlib.suppress(Exception):
            stream.close()
        self._remove_tmp()

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
        self._tmp_path = self.filename + ".tmp"
        with _translate_io_error(self._tmp_path):
            self._stream = _core.OFrameFStream(self._tmp_path)

    # A reopened writer starts a new file, with a fresh channel set
    self._channel_names = None

    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, aux_mask: bool | dict[str, bool] = False)

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'`
`aux_mask`	`bool or dict`	If True, preserve partial masks on ADC channels in an auxiliary FrVect instead of the whole-channel dataValid flag (default: False). A dict maps channel names to bools. See `Frame.add_channel`.	`False`

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,
    aux_mask: bool | dict[str, bool] = False,
):
    """
    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)
    aux_mask : bool or dict, optional
        If True, preserve partial masks on ADC channels in an auxiliary
        FrVect instead of the whole-channel dataValid flag (default:
        False). A dict maps channel names to bools. See
        ``Frame.add_channel``.

    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,
        aux_mask=aux_mask,
    )

    # 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.

Raises:

Type	Description
`ValueError`	If the frame's channel set differs from that of the frames already written. The GWF table of contents stores one position per frame for every channel, so all frames in a file must contain the same channels; a mismatch would silently produce an unreadable file.

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.

    Raises
    ------
    ValueError
        If the frame's channel set differs from that of the frames already
        written. The GWF table of contents stores one position per frame
        for every channel, so all frames in a file must contain the same
        channels; a mismatch would silently produce an unreadable file.
    """
    if self._stream is None:
        msg = "FrameWriter not opened (call open() or use 'with' statement)"
        raise RuntimeError(msg)

    channel_names = frozenset(frame)
    with self._channel_names_lock:
        if self._channel_names is None:
            self._channel_names = channel_names
        elif channel_names != self._channel_names:
            msg = (
                f"frame channel set {sorted(channel_names)} does not match "
                f"this file's channel set {sorted(self._channel_names)}: "
                "all frames in a GWF file must contain the same channels"
            )
            raise ValueError(msg)

    with _translate_io_error(self._tmp_path):
        frame.write_to_stream(
            self._stream, self.compression, self.compression_level
        )

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

write ¶

write(filename: str | PathLike[str], 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, aux_mask: bool | dict[str, bool] = False, frame_spec: int | None = None)

Write channel data to a GWF file.

This is a convenience function for simple write operations. For more control, use the Frame class directly.

Parameters:

Name	Type	Description	Default
`filename`	`str or path - like`	Output file path	required
`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'`
`compression`	`int`	Compression scheme (default: Compression.ZERO_SUPPRESS_OTHERWISE_GZIP)	`ZERO_SUPPRESS_OTHERWISE_GZIP`
`compression_level`	`int`	Compression level 0-9 (default: 6)	`6`
`aux_mask`	`bool or dict`	If True, preserve partial masks on ADC channels in an auxiliary FrVect instead of the whole-channel dataValid flag (default: False). A dict maps channel names to bools. See `Frame.add_channel`.	`False`

Examples:

>>> # Write single channel
>>> data = np.sin(np.linspace(0, 2*np.pi, 16384))
>>> gwframe.write('output.gwf', data, start=1234567890.0, sample_rate=16384,
...               name='L1:TEST', unit='strain')

>>> # Write multiple channels
>>> gwframe.write('output.gwf',
...               channels={'L1:CHAN1': data1, 'L1:CHAN2': data2},
...               start=1234567890.0, sample_rate=16384, name='L1')

>>> # Write with different sample rates
>>> gwframe.write('output.gwf',
...               channels={'L1:FAST': data1, 'L1:SLOW': data2},
...               start=1234567890.0,
...               sample_rate={'L1:FAST': 16384, 'L1:SLOW': 256},
...               name='L1')

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, aux_mask: bool | dict[str, bool] = False, 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')

write

Frame ¶

frame_number property writable ¶

frame_spec property ¶

start property writable ¶

__delitem__ ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__setitem__ ¶

add_channel ¶

add_history ¶

write ¶

write_bytes ¶

write_to_stream ¶

FrameWriter ¶

close ¶

discard ¶

open ¶

write ¶

write_frame ¶

write ¶

write_bytes ¶

frame_number `property` `writable` ¶

frame_spec `property` ¶

start `property` `writable` ¶

delitem ¶

getitem ¶

iter ¶

len ¶

setitem ¶