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

    # 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

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, 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`

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, 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 ¶

start property writable ¶

__delitem__ ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__setitem__ ¶

add_channel ¶

add_history ¶

write ¶

write_bytes ¶

write_to_stream ¶

FrameWriter ¶

close ¶

open ¶

write ¶

write_frame ¶

write ¶

write_bytes ¶

start `property` `writable` ¶

delitem ¶

getitem ¶

iter ¶

len ¶

setitem ¶