WDF File I/O / WDFファイル入出力

The wandas.io.wdf_io module provides functionality for saving and loading ChannelFrame objects in the WDF (Wandas Data File) format. wandas.io.wdf_io モジュールは、ChannelFrame オブジェクトを WDF (Wandas Data File) 形式で保存・読み込みするための機能を提供します。

The WDF format is based on HDF5 and preserves not only the data but also all metadata such as sampling rate, units, and channel labels. WDFフォーマットは HDF5 をベースとし、データだけでなくサンプリングレート、単位、チャンネルラベルなどのメタデータも完全に保存します。

WDF Format Overview / WDFフォーマット概要

The WDF format has the following features: WDFフォーマットは以下の特徴を持ちます:

• HDF5-based hierarchical data structure. HDF5ベースの階層的なデータ構造。
• Complete preservation of channel data and metadata. チャンネルデータとメタデータの完全な保持。
• Size optimization through data compression and chunking. データ圧縮とチャンク化によるサイズ最適化。
• Version management for future extensions. 将来の拡張に対応するバージョン管理。

File structure / ファイル構造:

/meta           : Frame-level metadata (JSON format) / Frame 全体のメタデータ (JSON形式)
/channels/{i}   : Individual channel data and metadata / 個々のチャンネルデータとメタデータ
    ├─ data           : Waveform data (numpy array) / 波形データ (numpy array)
    └─ attrs          : Channel attributes (labels, units, etc.) / チャンネル属性 (ラベル、単位など)

Saving WDF Files / WDFファイル保存

wandas.io.wdf_io.save(frame, path, *, format='hdf5', compress='gzip', overwrite=False, dtype=None)

Save a frame to a file.

Parameters:

Name	Type	Description	Default
frame	BaseFrame[Any]	The frame to save.	required
path	str | Path	Path to save the file. '.wdf' extension will be added if not present.	required
format	str	Format to use (currently only 'hdf5' is supported)	'hdf5'
compress	str | None	Compression method ('gzip' by default, None for no compression)	'gzip'
overwrite	bool	Whether to overwrite existing file	False
dtype	str | dtype[Any] | None	Optional data type conversion before saving (e.g. 'float32')	None

Raises:

Type	Description
FileExistsError	If the file exists and overwrite=False.
NotImplementedError	For unsupported formats.

Source code in wandas/io/wdf_io.py

def save(
    frame: BaseFrame[Any],
    path: str | Path,
    *,
    format: str = "hdf5",
    compress: str | None = "gzip",
    overwrite: bool = False,
    dtype: str | np.dtype[Any] | None = None,
) -> None:
    """Save a frame to a file.

    Args:
        frame: The frame to save.
        path: Path to save the file. '.wdf' extension will be added if not present.
        format: Format to use (currently only 'hdf5' is supported)
        compress: Compression method ('gzip' by default, None for no compression)
        overwrite: Whether to overwrite existing file
        dtype: Optional data type conversion before saving (e.g. 'float32')

    Raises:
        FileExistsError: If the file exists and overwrite=False.
        NotImplementedError: For unsupported formats.
    """
    # Handle path
    path = Path(path)
    if path.suffix != ".wdf":
        path = path.with_suffix(".wdf")

    # Check if file exists
    if path.exists() and not overwrite:
        raise FileExistsError(f"File {path} already exists. Set overwrite=True to overwrite.")

    # Currently only HDF5 is supported
    if format.lower() != "hdf5":
        raise NotImplementedError(f"Format {format} not supported. Only 'hdf5' is currently implemented.")

    # Compute data arrays (this triggers actual computation)
    logger.info("Computing data arrays for saving...")
    computed_data = frame.compute()
    if dtype is not None:
        computed_data = computed_data.astype(dtype)

    # Create file
    logger.info(f"Creating HDF5 file at {path}...")
    with h5py.File(path, "w") as f:
        # Set file version
        f.attrs["version"] = WDF_FORMAT_VERSION

        # Store frame metadata
        f.attrs["sampling_rate"] = frame.sampling_rate
        f.attrs["label"] = frame.label or ""
        f.attrs["frame_type"] = type(frame).__name__

        # Create channels group
        channels_grp = f.create_group("channels")

        # Store each channel
        for i, (channel_data, ch_meta) in enumerate(zip(computed_data, frame._channel_metadata)):
            ch_grp = channels_grp.create_group(f"{i}")

            # Store channel data
            if compress:
                ch_grp.create_dataset("data", data=channel_data, compression=compress)
            else:
                ch_grp.create_dataset("data", data=channel_data)

            # Store metadata
            ch_grp.attrs["label"] = ch_meta.label
            ch_grp.attrs["unit"] = ch_meta.unit

            # Store extra metadata as JSON
            if ch_meta.extra:
                ch_grp.attrs["metadata_json"] = json.dumps(ch_meta.extra)

        # Store operation history
        if frame.operation_history:
            op_grp = f.create_group("operation_history")
            for i, op in enumerate(frame.operation_history):
                op_sub_grp = op_grp.create_group(f"operation_{i}")
                for k, v in op.items():
                    # Store simple attributes directly
                    if isinstance(v, (str, int, float, bool, np.number)):
                        op_sub_grp.attrs[k] = v
                    else:
                        # For complex types, serialize to JSON
                        try:
                            op_sub_grp.attrs[k] = json.dumps(v)
                        except (TypeError, OverflowError) as e:
                            logger.warning(f"Could not serialize operation key '{k}': {e}")
                            op_sub_grp.attrs[k] = str(v)

        # Store frame metadata
        dict_is_nonempty = bool(frame.metadata)
        has_source_file = isinstance(frame.metadata, FrameMetadata) and frame.metadata.source_file is not None
        if dict_is_nonempty or has_source_file:
            meta_grp = f.create_group("meta")
            # Store metadata dict content as JSON
            meta_grp.attrs["json"] = json.dumps(dict(frame.metadata))

            # Store source_file separately if present
            if has_source_file:
                meta_grp.attrs["source_file"] = str(frame.metadata.source_file)

            # Also store individual metadata items as attributes for compatibility
            for k, v in frame.metadata.items():
                if isinstance(v, (str, int, float, bool, np.number)):
                    meta_grp.attrs[k] = v

    logger.info(f"Frame saved to {path}")

Loading WDF Files / WDFファイル読み込み

wandas.io.wdf_io.load(path, *, format='hdf5', timeout=10.0)

Load a ChannelFrame object from a WDF (Wandas Data File) file or URL.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the WDF file to load, or an HTTP/HTTPS URL pointing to a remote WDF file. When a URL is given the file is downloaded in full before opening.	required
format	str	Format of the file. Currently only "hdf5" is supported.	'hdf5'
timeout	float	Timeout in seconds for HTTP/HTTPS URL downloads. Default is 10.0 seconds. Has no effect for local file paths.	10.0

Returns:

Type	Description
ChannelFrame	A new ChannelFrame object with data and metadata loaded from the file.

Raises:

Type	Description
FileNotFoundError	If the file doesn't exist.
NotImplementedError	If format is not "hdf5".
ValueError	If the file format is invalid or incompatible.

Example

cf = ChannelFrame.load("audio_data.wdf") cf = ChannelFrame.load("https://example.com/audio_data.wdf")

Source code in wandas/io/wdf_io.py

def load(path: str | Path, *, format: str = "hdf5", timeout: float = 10.0) -> "ChannelFrame":
    """Load a ChannelFrame object from a WDF (Wandas Data File) file or URL.

    Args:
        path: Path to the WDF file to load, or an HTTP/HTTPS URL pointing to
            a remote WDF file. When a URL is given the file is downloaded in
            full before opening.
        format: Format of the file. Currently only "hdf5" is supported.
        timeout: Timeout in seconds for HTTP/HTTPS URL downloads. Default is
            10.0 seconds. Has no effect for local file paths.

    Returns:
        A new ChannelFrame object with data and metadata loaded from the file.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        NotImplementedError: If format is not "hdf5".
        ValueError: If the file format is invalid or incompatible.

    Example:
        >>> cf = ChannelFrame.load("audio_data.wdf")
        >>> cf = ChannelFrame.load("https://example.com/audio_data.wdf")
    """
    # Ensure ChannelFrame is imported here to avoid circular imports
    from ..core.metadata import ChannelMetadata
    from ..frames.channel import ChannelFrame

    if format.lower() != "hdf5":
        raise NotImplementedError(f"Format '{format}' is not supported")

    # Detect and handle URL paths — download to memory before HDF5 open.
    h5_source: str | Path | io.BytesIO
    h5_kwargs: dict[str, object] = {}
    if isinstance(path, str) and (path.startswith("http://") or path.startswith("https://")):
        import urllib.error
        import urllib.request

        logger.debug(f"Downloading WDF from URL: {path}")
        try:
            with urllib.request.urlopen(path, timeout=timeout) as _resp:
                h5_source = io.BytesIO(_resp.read())
        except urllib.error.URLError as exc:
            raise OSError(
                f"Failed to download WDF file from URL\n"
                f"  URL: {path}\n"
                f"  Error: {exc}\n"
                f"Verify the URL is accessible and try again."
            ) from exc
        h5_kwargs = {"driver": "fileobj"}
    else:
        path = Path(path)
        if not path.exists():
            raise FileNotFoundError(f"File not found: {path}")
        h5_source = path

    logger.debug(f"Loading ChannelFrame from {h5_source!r}")

    with h5py.File(h5_source, "r", **h5_kwargs) as f:
        # Check format version for compatibility
        version = f.attrs.get("version", "unknown")
        if version != WDF_FORMAT_VERSION:
            logger.warning(
                f"File format version mismatch: file={version}, current={WDF_FORMAT_VERSION}"  # noqa: E501
            )

        # Get global attributes
        sampling_rate = float(f.attrs["sampling_rate"])
        frame_label = f.attrs.get("label", "")

        # Get frame metadata
        frame_metadata = FrameMetadata()
        if "meta" in f:
            meta_json = f["meta"].attrs.get("json", "{}")
            if isinstance(meta_json, (bytes, np.bytes_)):
                try:
                    meta_json = meta_json.decode("utf-8")
                except (UnicodeDecodeError, AttributeError):
                    meta_json = str(meta_json)
            frame_metadata.update(json.loads(meta_json))
            source_file = f["meta"].attrs.get("source_file", None)
            if source_file is not None:
                if isinstance(source_file, (bytes, np.bytes_)):
                    try:
                        source_file = source_file.decode("utf-8")
                    except (UnicodeDecodeError, AttributeError):
                        source_file = str(source_file)
                frame_metadata.source_file = str(source_file)

        # Load operation history
        operation_history = []
        if "operation_history" in f:
            op_grp = f["operation_history"]
            # Sort operation indices numerically
            op_indices = sorted([int(key.split("_")[1]) for key in op_grp.keys()])

            for idx in op_indices:
                op_sub_grp = op_grp[f"operation_{idx}"]
                op_dict = {}
                for attr_name in op_sub_grp.attrs:
                    attr_value = op_sub_grp.attrs[attr_name]
                    # Try to deserialize JSON, fallback to string
                    try:
                        op_dict[attr_name] = json.loads(attr_value)
                    except (json.JSONDecodeError, TypeError):
                        op_dict[attr_name] = attr_value
                operation_history.append(op_dict)

        # Load channel data and metadata
        all_channel_data = []
        channel_metadata_list = []

        if "channels" in f:
            channels_group = f["channels"]
            # Sort channel indices numerically
            channel_indices = sorted([int(key) for key in channels_group.keys()])

            for idx in channel_indices:
                ch_group = channels_group[f"{idx}"]

                # Load channel data
                channel_data = ch_group["data"][()]

                # Append to combined array
                all_channel_data.append(channel_data)

                # Load channel metadata
                label = ch_group.attrs.get("label", f"Ch{idx}")
                unit = ch_group.attrs.get("unit", "")

                # Load additional metadata if present
                ch_extra = {}
                if "metadata_json" in ch_group.attrs:
                    ch_extra = json.loads(ch_group.attrs["metadata_json"])

                # Create ChannelMetadata object
                channel_metadata = ChannelMetadata(label=label, unit=unit, extra=ch_extra)
                channel_metadata_list.append(channel_metadata)

        # Stack channel data into a single array
        if all_channel_data:
            combined_data = np.stack(all_channel_data, axis=0)
        else:
            raise ValueError("No channel data found in the file")

        # Create a new ChannelFrame
        # Use channel-wise chunking: 1 for channel axis and -1 for samples
        dask_data = _da_from_array(combined_data, chunks=(1, -1))

        cf = ChannelFrame(
            data=dask_data,
            sampling_rate=sampling_rate,
            label=frame_label if frame_label else None,
            metadata=frame_metadata,
            operation_history=operation_history,
            channel_metadata=channel_metadata_list,
        )

        logger.debug(
            f"ChannelFrame loaded from {path}: {len(cf)} channels, {cf.n_samples} samples"  # noqa: E501
        )
        return cf

Usage Examples / 利用例

# Save a ChannelFrame in WDF format
# ChannelFrame を WDF形式で保存
cf = wd.read_wav("audio.wav")
cf.save("audio_data.wdf")

# Specifying options when saving
# 保存時のオプション指定
cf.save(
    "high_quality.wdf",
    compress="gzip",  # Compression method / 圧縮方式
    dtype="float64",  # Data type / データ型
    overwrite=True    # Allow overwriting / 上書き許可
)

# Load a ChannelFrame from a WDF file
# WDFファイルから ChannelFrame を読み込み
cf2 = wd.ChannelFrame.load("audio_data.wdf")