IO Module / 入出力モジュール

The wandas.io module provides readers for external data and persistence for WDF artifacts. Input, output, units, and compatibility contracts live in the generated docstrings.

wandas.ioは外部データのreaderとWDF artifactの永続化を提供します。入力、出力、単位、 互換性の契約は生成されたdocstringを正本とします。

wandas.inspect(source, *, file_type=None, source_name=None)

Inspect an audio header before constructing a Frame or decoding samples.

This synchronous preflight supports the built-in SoundFile audio reader (WAV, W64, RF64, FLAC, OGG/OGA, Opus, MP3, AIFF/AIF, SND/AU and CAF). It neither constructs a Dask graph nor decodes PCM. CSV, WDF, URL downloads and custom readers are deliberately unsupported: CSV metadata requires parsing the sampled table, while custom readers cannot guarantee header-only inspection. Use read or load for those workflows.

In-memory format inference matches read: explicit file_type, then a stream's .name suffix, then source_name, then anonymous WAV. Local paths use their suffix unless file_type is explicit. A seekable binary stream is inspected from position zero and its original position is restored on success or failure. A borrowed stream is never closed. Local handles opened internally are closed by the reader. This is a snapshot; reading the source later inspects it again and does not reuse this result.

Parameters:

Name	Type	Description	Default
source	str | Path | bytes | bytearray | memoryview | BinaryIO	Local path, bytes-like value, or seekable readable binary stream.	required
file_type	str | None	Optional audio extension, case-insensitive with or without a dot.	None
source_name	str | None	Optional in-memory format hint; never another resource to open.	None

Returns:

Type	Description
dict[str, Any]	A dictionary with samplerate (Hz), channels, frames,
dict[str, Any]	duration (seconds), format, subtype and unit. Values
dict[str, Any]	describe the complete original recording, without channel or time slicing.

Raises:

Type	Description
TypeError	The source type is unsupported or the stream is text.
ValueError	The format is unsupported, a URL is supplied, or a stream cannot be rewound and restored without consuming input.
FileNotFoundError	A local path does not exist.
OSError	A filesystem operation fails for an inaccessible local source.
RuntimeError	SoundFile raises LibsndfileError for corrupt or unsupported audio; the exception propagates without wrapping.

Examples:

>>> import wandas as wd
>>> info = wd.inspect("recording.wav")
>>> if info["channels"] <= 8 and info["duration"] <= 180:
...     recording = wd.read("recording.wav")

Source code in wandas/io/inspect.py

def inspect(
    source: str | Path | bytes | bytearray | memoryview | BinaryIO,
    *,
    file_type: str | None = None,
    source_name: str | None = None,
) -> dict[str, Any]:
    """Inspect an audio header before constructing a Frame or decoding samples.

    This synchronous preflight supports the built-in SoundFile audio reader
    (WAV, W64, RF64, FLAC, OGG/OGA, Opus, MP3, AIFF/AIF, SND/AU and CAF). It neither constructs a Dask graph nor
    decodes PCM. CSV, WDF, URL downloads and custom readers are deliberately
    unsupported: CSV metadata requires parsing the sampled table, while custom
    readers cannot guarantee header-only inspection. Use ``read`` or ``load``
    for those workflows.

    In-memory format inference matches ``read``: explicit ``file_type``, then
    a stream's ``.name`` suffix, then ``source_name``, then anonymous WAV.
    Local paths use their suffix unless ``file_type`` is explicit. A seekable
    binary stream is inspected from position zero and its original position is
    restored on success or failure. A borrowed stream is never closed. Local
    handles opened internally are closed by the reader. This is a snapshot;
    reading the source later inspects it again and does not reuse this result.

    Args:
        source: Local path, bytes-like value, or seekable readable binary stream.
        file_type: Optional audio extension, case-insensitive with or without a dot.
        source_name: Optional in-memory format hint; never another resource to open.

    Returns:
        A dictionary with ``samplerate`` (Hz), ``channels``, ``frames``,
        ``duration`` (seconds), ``format``, ``subtype`` and ``unit``. Values
        describe the complete original recording, without channel or time slicing.

    Raises:
        TypeError: The source type is unsupported or the stream is text.
        ValueError: The format is unsupported, a URL is supplied, or a stream
            cannot be rewound and restored without consuming input.
        FileNotFoundError: A local path does not exist.
        OSError: A filesystem operation fails for an inaccessible local source.
        RuntimeError: SoundFile raises ``LibsndfileError`` for corrupt or
            unsupported audio; the exception propagates without wrapping.

    Examples:
        >>> import wandas as wd
        >>> info = wd.inspect("recording.wav")
        >>> if info["channels"] <= 8 and info["duration"] <= 180:
        ...     recording = wd.read("recording.wav")
    """
    if isinstance(source, io.TextIOBase):
        raise TypeError("inspect requires a binary stream, not a text stream")
    if isinstance(source, (str, Path)):
        if "://" in str(source):
            raise ValueError("inspect accepts local audio sources, not URL downloads")
        if not Path(source).exists():
            raise FileNotFoundError(source)
    elif not isinstance(source, (bytes, bytearray, memoryview)) and not _is_file_like(source):
        raise TypeError("inspect requires a local path, bytes or a seekable binary stream")
    selected_type = _infer_in_memory_file_type(source, file_type, source_name)
    reader = get_file_reader(source, file_type=selected_type)
    if type(reader) is not SoundFileReader:
        raise ValueError("inspect supports built-in audio headers only; use read() for CSV or custom readers")
    if not _is_file_like(source):
        return reader.get_file_info(source)
    stream = cast(BinaryIO, source)
    try:
        if not stream.seekable():
            raise ValueError("stream is not seekable")
        position = stream.tell()
        stream.seek(position)
    except (AttributeError, OSError, ValueError) as error:
        raise ValueError("inspect requires a seekable binary stream with a restorable position") from error
    try:
        try:
            stream.seek(0)
        except (OSError, ValueError) as error:
            raise ValueError("inspect requires a seekable binary stream that can rewind to position zero") from error
        return reader.get_file_info(stream)
    finally:
        stream.seek(position)

wandas.io.readers

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

DownloadedTemporaryFile dataclass

Temporary file created for streamed URL downloads.

Source code in wandas/io/readers.py

@dataclass
class DownloadedTemporaryFile:
    """Temporary file created for streamed URL downloads."""

    path: Path
    temp_dir: tempfile.TemporaryDirectory[str]

    def __post_init__(self) -> None:
        self._finalizer = weakref.finalize(
            self,
            tempfile.TemporaryDirectory.cleanup,
            self.temp_dir,
        )

    def __enter__(self) -> "DownloadedTemporaryFile":
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        self.cleanup()

    def cleanup(self) -> None:
        if self._finalizer.alive:
            self._finalizer()

Attributes

path instance-attribute

temp_dir instance-attribute

Functions

__init__(path, temp_dir)

__post_init__()

Source code in wandas/io/readers.py

def __post_init__(self) -> None:
    self._finalizer = weakref.finalize(
        self,
        tempfile.TemporaryDirectory.cleanup,
        self.temp_dir,
    )

__enter__()

Source code in wandas/io/readers.py

def __enter__(self) -> "DownloadedTemporaryFile":
    return self

__exit__(exc_type, exc, tb)

Source code in wandas/io/readers.py

def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
    self.cleanup()

cleanup()

Source code in wandas/io/readers.py

def cleanup(self) -> None:
    if self._finalizer.alive:
        self._finalizer()

CSVFileInfoParams

Bases: TypedDict

Type definition for CSV file reader parameters in get_file_info.

Attributes:

Name	Type	Description
delimiter	str	Delimiter character. Defaults to ",".
header	Optional[int]	Row number to use as header. Defaults to 0 (first row); use None if there is no header.
time_column	Union[int, str]	Index or name of the time column. Defaults to 0.

Source code in wandas/io/readers.py

class CSVFileInfoParams(TypedDict, total=False):
    """Type definition for CSV file reader parameters in ``get_file_info``.

    Attributes:
        delimiter (str): Delimiter character. Defaults to `","`.
        header (Optional[int]): Row number to use as header. Defaults to 0
            (first row); use ``None`` if there is no header.
        time_column (Union[int, str]): Index or name of the time column.
            Defaults to 0.
    """

    delimiter: str
    header: int | None
    time_column: int | str

Attributes

delimiter instance-attribute

header instance-attribute

time_column instance-attribute

CSVGetDataParams

Bases: TypedDict

Type definition for CSV file reader parameters in get_data.

Attributes:

Name	Type	Description
delimiter	str	Delimiter character. Defaults to ",".
header	Optional[int]	Row number to use as header. Defaults to 0 (first row); use None if there is no header.
time_column	Union[int, str]	Index or name of the time column. Defaults to 0.

Source code in wandas/io/readers.py

class CSVGetDataParams(TypedDict, total=False):
    """Type definition for CSV file reader parameters in ``get_data``.

    Attributes:
        delimiter (str): Delimiter character. Defaults to `","`.
        header (Optional[int]): Row number to use as header. Defaults to 0
            (first row); use ``None`` if there is no header.
        time_column (Union[int, str]): Index or name of the time column.
            Defaults to 0.
    """

    delimiter: str
    header: int | None
    time_column: int | str

Attributes

delimiter instance-attribute

header instance-attribute

time_column instance-attribute

FileReader

Bases: ABC

Base class for external data readers.

Implementations return real channel-first arrays. Audio readers must apply full-scale decoding before returning values.

Source code in wandas/io/readers.py

class FileReader(ABC):
    """Base class for external data readers.

    Implementations return real channel-first arrays. Audio readers must apply
    full-scale decoding before returning values.
    """

    # Class attribute for supported file extensions
    supported_extensions: ClassVar[list[str]] = []

    @classmethod
    @abstractmethod
    def get_file_info(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """Get basic information about the audio file.

        Args:
            path: Path to the file.
            **kwargs: Additional parameters specific to the file reader.

        Returns:
            Dictionary containing file information including:
                - samplerate: Sampling rate in Hz
                - channels: Number of channels
                - frames: Total number of frames
                - format: File format
                - duration: Duration in seconds
        """
        # pragma: no cover

    @classmethod
    @abstractmethod
    def get_data(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        channels: list[int],
        start_idx: int,
        frames: int,
        **kwargs: Any,
    ) -> ArrayLike:
        """Read audio data from the file.

        Args:
            path: Path to the file.
            channels: List of channel indices to read.
            start_idx: Starting frame index.
            frames: Number of frames to read.
            **kwargs: Additional parameters specific to the file reader.

        Returns:
            Array of shape (channels, frames) containing the audio data.
        """
        # pragma: no cover

    @classmethod
    def _normalize_supported_extensions(cls) -> list[str]:
        """Return normalized reader extensions in lowercase dot form."""
        normalized_extensions: list[str] = []
        for extension in cls.supported_extensions:
            normalized_extension = _normalize_extension(extension)
            if normalized_extension is not None:
                normalized_extensions.append(normalized_extension)
        return normalized_extensions

    @classmethod
    def can_read(cls, path: str | Path) -> bool:
        """Check if this reader can handle the file based on extension."""
        ext = _normalize_extension(Path(path).suffix)
        if ext is None:
            return False
        return ext in cls._normalize_supported_extensions()

Attributes

supported_extensions = [] class-attribute

Functions

get_file_info(path, **kwargs) abstractmethod classmethod

Get basic information about the audio file.

Parameters:

Name	Type	Description	Default
path	str | Path | bytes | bytearray | memoryview | BinaryIO	Path to the file.	required
**kwargs	Any	Additional parameters specific to the file reader.	{}

Returns:

Type	Description
dict[str, Any]	Dictionary containing file information including: - samplerate: Sampling rate in Hz - channels: Number of channels - frames: Total number of frames - format: File format - duration: Duration in seconds

Source code in wandas/io/readers.py

@classmethod
@abstractmethod
def get_file_info(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    **kwargs: Any,
) -> dict[str, Any]:
    """Get basic information about the audio file.

    Args:
        path: Path to the file.
        **kwargs: Additional parameters specific to the file reader.

    Returns:
        Dictionary containing file information including:
            - samplerate: Sampling rate in Hz
            - channels: Number of channels
            - frames: Total number of frames
            - format: File format
            - duration: Duration in seconds
    """

get_data(path, channels, start_idx, frames, **kwargs) abstractmethod classmethod

Read audio data from the file.

Parameters:

Name	Type	Description	Default
path	str | Path | bytes | bytearray | memoryview | BinaryIO	Path to the file.	required
channels	list[int]	List of channel indices to read.	required
start_idx	int	Starting frame index.	required
frames	int	Number of frames to read.	required
**kwargs	Any	Additional parameters specific to the file reader.	{}

Returns:

Type	Description
ArrayLike	Array of shape (channels, frames) containing the audio data.

Source code in wandas/io/readers.py

@classmethod
@abstractmethod
def get_data(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    channels: list[int],
    start_idx: int,
    frames: int,
    **kwargs: Any,
) -> ArrayLike:
    """Read audio data from the file.

    Args:
        path: Path to the file.
        channels: List of channel indices to read.
        start_idx: Starting frame index.
        frames: Number of frames to read.
        **kwargs: Additional parameters specific to the file reader.

    Returns:
        Array of shape (channels, frames) containing the audio data.
    """

can_read(path) classmethod

Check if this reader can handle the file based on extension.

Source code in wandas/io/readers.py

@classmethod
def can_read(cls, path: str | Path) -> bool:
    """Check if this reader can handle the file based on extension."""
    ext = _normalize_extension(Path(path).suffix)
    if ext is None:
        return False
    return ext in cls._normalize_supported_extensions()

SoundFileReader

Bases: FileReader

Audio file reader using SoundFile library.

Source code in wandas/io/readers.py

class SoundFileReader(FileReader):
    """Audio file reader using SoundFile library."""

    # libsndfile detects the container from the header, so these suffixes only route files here.
    supported_extensions: ClassVar[list[str]] = _soundfile_extensions()

    @classmethod
    def get_file_info(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """Get basic information about the audio file."""
        info = sf.info(_prepare_file_source(path))
        return {
            "samplerate": info.samplerate,
            "channels": info.channels,
            "frames": info.frames,
            "format": info.format,
            "subtype": info.subtype,
            "duration": info.frames / info.samplerate,
            # The decoder contract is canonical full-scale float. Represent
            # that measurement domain through existing channel unit/ref
            # metadata rather than inferring it from identity calibration.
            "unit": "FS",
        }

    @classmethod
    def get_data(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        channels: list[int],
        start_idx: int,
        frames: int,
        **kwargs: Any,
    ) -> ArrayLike:
        """Read full-scale audio data from the file.

        Args:
            path: Path, bytes or binary stream of the audio file.
            channels: Channel indices to read.
            start_idx: First frame to read.
            frames: Number of frames, or a negative value for all remaining frames.
            **kwargs: ``dtype`` selects the decoded dtype, ``float32`` or ``float64``
                (the default). Other keys are ignored.
        """
        dtype = np.dtype(kwargs.get("dtype", np.float64))
        logger.debug(f"Reading {frames} frames from {path!r} starting at {start_idx}")

        with sf.SoundFile(_prepare_file_source(path)) as f:
            if start_idx > 0:
                f.seek(start_idx)
            if f.seekable():
                # Clamp before allocating, as SoundFile.read does when it allocates.
                remaining = max(f.frames - f.tell(), 0)
                frames = remaining if frames < 0 else min(frames, remaining)
            # Decode in fixed-size blocks and copy only the requested channels,
            # so peak memory is the selected output plus one all-channel block
            # instead of the whole file in every channel.
            result = np.empty((len(channels), frames), dtype=dtype)
            block = np.empty((min(frames, _SOUNDFILE_READ_BLOCK_FRAMES), f.channels), dtype=dtype)
            filled = 0
            while filled < frames:
                read = f.read(frames=min(frames - filled, len(block)), dtype=dtype.name, always_2d=True, out=block)
                if len(read) == 0:
                    break
                result[:, filled : filled + len(read)] = read[:, channels].T
                filled += len(read)
            if filled < frames:
                result = result[:, :filled]

        _shape = result.shape
        logger.debug(f"File read complete, returning data with shape {_shape}")
        return result

Attributes

supported_extensions = _soundfile_extensions() class-attribute

Functions

get_file_info(path, **kwargs) classmethod

Get basic information about the audio file.

Source code in wandas/io/readers.py

@classmethod
def get_file_info(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    **kwargs: Any,
) -> dict[str, Any]:
    """Get basic information about the audio file."""
    info = sf.info(_prepare_file_source(path))
    return {
        "samplerate": info.samplerate,
        "channels": info.channels,
        "frames": info.frames,
        "format": info.format,
        "subtype": info.subtype,
        "duration": info.frames / info.samplerate,
        # The decoder contract is canonical full-scale float. Represent
        # that measurement domain through existing channel unit/ref
        # metadata rather than inferring it from identity calibration.
        "unit": "FS",
    }

get_data(path, channels, start_idx, frames, **kwargs) classmethod

Read full-scale audio data from the file.

Parameters:

Name	Type	Description	Default
path	str | Path | bytes | bytearray | memoryview | BinaryIO	Path, bytes or binary stream of the audio file.	required
channels	list[int]	Channel indices to read.	required
start_idx	int	First frame to read.	required
frames	int	Number of frames, or a negative value for all remaining frames.	required
**kwargs	Any	dtype selects the decoded dtype, float32 or float64 (the default). Other keys are ignored.	{}

Source code in wandas/io/readers.py

@classmethod
def get_data(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    channels: list[int],
    start_idx: int,
    frames: int,
    **kwargs: Any,
) -> ArrayLike:
    """Read full-scale audio data from the file.

    Args:
        path: Path, bytes or binary stream of the audio file.
        channels: Channel indices to read.
        start_idx: First frame to read.
        frames: Number of frames, or a negative value for all remaining frames.
        **kwargs: ``dtype`` selects the decoded dtype, ``float32`` or ``float64``
            (the default). Other keys are ignored.
    """
    dtype = np.dtype(kwargs.get("dtype", np.float64))
    logger.debug(f"Reading {frames} frames from {path!r} starting at {start_idx}")

    with sf.SoundFile(_prepare_file_source(path)) as f:
        if start_idx > 0:
            f.seek(start_idx)
        if f.seekable():
            # Clamp before allocating, as SoundFile.read does when it allocates.
            remaining = max(f.frames - f.tell(), 0)
            frames = remaining if frames < 0 else min(frames, remaining)
        # Decode in fixed-size blocks and copy only the requested channels,
        # so peak memory is the selected output plus one all-channel block
        # instead of the whole file in every channel.
        result = np.empty((len(channels), frames), dtype=dtype)
        block = np.empty((min(frames, _SOUNDFILE_READ_BLOCK_FRAMES), f.channels), dtype=dtype)
        filled = 0
        while filled < frames:
            read = f.read(frames=min(frames - filled, len(block)), dtype=dtype.name, always_2d=True, out=block)
            if len(read) == 0:
                break
            result[:, filled : filled + len(read)] = read[:, channels].T
            filled += len(read)
        if filled < frames:
            result = result[:, :filled]

    _shape = result.shape
    logger.debug(f"File read complete, returning data with shape {_shape}")
    return result

CSVFileReader

Bases: FileReader

CSV file reader for time series data.

Metadata inspection eagerly parses the complete table to determine its exact shape and sampling rate. Sample values in the public Frame remain Dask-backed, and get_data() parses the table again when that graph is computed. Rates are estimated from the mean positive time interval without rounding fractional rates to integers. Only integer-near floating-point roundoff is corrected. Unusable time coordinates report rate zero, which the public Frame reader rejects before indexing samples. Increasing times that deviate by more than a quarter of the mean interval from the uniform grid starting at the first time raise ValueError. When the times show the pattern of rounding to a decimal step, they are instead compared with their best-fitting uniform grid and may deviate by up to half that step, so times printed with few decimals are accepted. When the printed intervals are one and two steps, the two-step interval must occur at least twice for the times to count as rounded, because a single one marks a dropped sample in exact times; a short rounded file with only one (for example 20 rows of 96 kHz times printed with 5 decimals) is rejected. Both limits add a few float64 spacings at the times' magnitude. When that step or float64 resolution approaches an interval, some dropped samples match a rounded grid of a slightly lower rate and go undetected; for example, a 1 kHz logger with millisecond times and two evenly spaced missing rows reads as about 998 Hz.

Source code in wandas/io/readers.py

class CSVFileReader(FileReader):
    """CSV file reader for time series data.

    Metadata inspection eagerly parses the complete table to determine its exact
    shape and sampling rate. Sample values in the public Frame remain Dask-backed,
    and ``get_data()`` parses the table again when that graph is computed.
    Rates are estimated from the mean positive time interval without rounding
    fractional rates to integers. Only integer-near floating-point roundoff is
    corrected. Unusable time coordinates report rate zero, which the public
    Frame reader rejects before indexing samples. Increasing times that deviate
    by more than a quarter of the mean interval from the uniform grid starting
    at the first time raise ``ValueError``. When the times show the pattern of
    rounding to a decimal step, they are instead compared with their
    best-fitting uniform grid and may deviate by up to half that step, so times
    printed with few decimals are accepted. When the printed intervals are one
    and two steps, the two-step interval must occur at least twice for the times
    to count as rounded, because a single one marks a dropped sample in exact
    times; a short rounded file with only one (for example 20 rows of 96 kHz
    times printed with 5 decimals) is rejected. Both limits add a few float64
    spacings at the times' magnitude. When that step or float64 resolution
    approaches an interval, some dropped samples match a rounded grid of a
    slightly lower rate and go undetected; for example, a 1 kHz logger with
    millisecond times and two evenly spaced missing rows reads as about 998 Hz.
    """

    # CSV supported formats
    supported_extensions: ClassVar[list[str]] = [".csv"]

    @classmethod
    def get_file_info(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """Get basic information about the CSV file.

        Args:
            path: Union[str, Path]. Path to the CSV file.
            **kwargs: Any. Additional parameters for CSV reading. Supported parameters:

                - delimiter : str, default=","
                Delimiter character.
                - header : Optional[int], default=0
                Row number to use as header. Set to None if no header.
                - time_column : Union[int, str], default=0
                Index or name of the time column.

        Returns:
            dict[str, Any]: Dictionary containing file information including:
                - samplerate: Estimated sampling rate in Hz
                - channels: Number of data channels (excluding time column)
                - frames: Total number of frames
                - format: "CSV"
                - duration: Duration in seconds (or None if cannot be calculated)
                - ch_labels: List of channel labels

        Notes:
            This method accepts CSV-specific parameters through kwargs.
            See CSVFileInfoParams for supported parameter types.
        """
        # Extract parameters with defaults
        delimiter: str = kwargs.get("delimiter", ",")
        header: int | None = kwargs.get("header", 0)
        time_column: int | str = kwargs.get("time_column", 0)

        # Parse the complete table because exact frame count and sampling rate are
        # required before the public Dask array can be constructed.
        pd = require_pandas("CSV file reading")
        df = pd.read_csv(_prepare_file_source(path), delimiter=delimiter, header=header)
        time_index = _resolve_csv_time_column(df.columns.tolist(), time_column)

        # Estimate sampling rate from the selected time column.
        grid_deviation = 0.0
        allowed_deviation = 0.0
        mean_interval = 0.0
        try:
            time_series = df.iloc[:, time_index]
            time_values = np.asarray(time_series.values, dtype=np.float64)
            time_start = float(time_values[0]) if len(time_values) > 0 else 0.0
            intervals = np.diff(time_values)
            if len(time_values) > 1 and np.all(np.isfinite(time_values)) and np.all(intervals > 0):
                mean_interval = float(np.mean(intervals))
                float_tolerance = 4 * float(np.spacing(np.max(np.abs(time_values))))
                rounding_allowance = _printed_rounding_allowance(time_values, intervals, float_tolerance)
                if rounding_allowance > 0:
                    # Rounded times: accept exactly what rounding a uniform grid produces.
                    grid_deviation = _uniform_grid_deviation(time_values)
                    allowed_deviation = (
                        max(_CSV_MAX_GRID_DEVIATION * mean_interval, rounding_allowance)
                        + float_tolerance
                        + _GRID_SEARCH_TOLERANCE * mean_interval
                    )
                else:
                    # Exact times: anchor the grid at the first time, which keeps a
                    # dropped sample detectable even in a file of a few rows.
                    grid = time_start + mean_interval * np.arange(len(time_values))
                    grid_deviation = float(np.max(np.abs(time_values - grid)))
                    allowed_deviation = _CSV_MAX_GRID_DEVIATION * mean_interval + float_tolerance
                estimated_sr = float(1 / mean_interval)
                # Correct only integer-near roundoff; fractional rates are valid.
                nearest_integer = round(estimated_sr)
                if nearest_integer > 0 and np.isclose(estimated_sr, nearest_integer, rtol=1e-12, atol=0):
                    estimated_sr = float(nearest_integer)
            else:
                estimated_sr = 0.0  # Cannot infer a valid rate from these coordinates
        except Exception:
            estimated_sr = 0.0  # Default if can't calculate
            time_start = 0.0

        if grid_deviation > allowed_deviation:
            raise ValueError(
                "Irregular CSV time column\n"
                f"  Got: a time value {grid_deviation / mean_interval:.3g} sample intervals away from a uniform grid\n"
                f"  Expected: at most {allowed_deviation / mean_interval:.3g} intervals (a quarter interval, or\n"
                "    half the printed time step when the times are rounded, plus float64 resolution)\n"
                "Wandas Frames assume a constant sampling rate, so irregular times would be\n"
                "relabelled silently. Resample the data to a uniform time grid before loading,\n"
                "or write the times with more decimals if they were rounded when saved."
            )

        channel_labels = [str(column) for index, column in enumerate(df.columns) if index != time_index]
        frames = df.shape[0]
        duration = frames / estimated_sr if estimated_sr else None

        # Return file info
        return {
            "samplerate": estimated_sr,
            "channels": len(channel_labels),
            "frames": frames,
            "format": "CSV",
            "duration": duration,
            "ch_labels": channel_labels,
            "time_start": time_start,
        }

    @classmethod
    def get_data(
        cls,
        path: str | Path | bytes | bytearray | memoryview | BinaryIO,
        channels: list[int],
        start_idx: int,
        frames: int,
        **kwargs: Any,
    ) -> ArrayLike:
        """Read data from the CSV file.

        Args:
            path: Union[str, Path]. Path to the CSV file.
            channels: list[int]. List of channel indices to read.
            start_idx: int. Starting frame index.
            frames: int. Number of frames to read.
            **kwargs: Any. Additional parameters for CSV reading. Supported parameters:

                - delimiter : str, default=","
                Delimiter character.
                - header : Optional[int], default=0
                Row number to use as header.
                - time_column : Union[int, str], default=0
                Index or name of the time column.

        Returns:
            ArrayLike: Array of shape (channels, frames) containing the data.

        Notes:
            This method accepts CSV-specific parameters through kwargs.
            See CSVGetDataParams for supported parameter types.
        """
        # Extract parameters with defaults
        time_column: int | str = kwargs.get("time_column", 0)
        delimiter: str = kwargs.get("delimiter", ",")
        header: int | None = kwargs.get("header", 0)

        logger.debug(f"Reading CSV data from {path!r} starting at {start_idx}")

        # Read the CSV file
        pd = require_pandas("CSV file reading")
        df = pd.read_csv(_prepare_file_source(path), delimiter=delimiter, header=header)
        time_index = _resolve_csv_time_column(df.columns.tolist(), time_column)

        # Remove time column
        df = df.iloc[:, [index for index in range(df.shape[1]) if index != time_index]]

        # Select requested channels - adjust indices to account for time column removal
        if channels:
            try:
                data_df = df.iloc[:, channels]
            except IndexError as e:
                raise ValueError(f"Requested channels {channels} out of range") from e
        else:
            data_df = df

        # Handle start_idx and frames for partial reading
        end_idx = start_idx + frames if frames > 0 else None
        data_df = data_df.iloc[start_idx:end_idx]

        # Convert to numpy array and transpose to (channels, samples) format
        try:
            result = data_df.to_numpy(dtype=np.float64).T
        except (AttributeError, TypeError, ValueError) as exc:
            non_numeric = [
                str(column) for column in data_df.columns if not pd.api.types.is_numeric_dtype(data_df[column])
            ]
            if not non_numeric:
                raise ValueError("Unexpected data type after reading file") from exc
            raise ValueError(
                "CSV data channels must be numeric\n"
                f"  Non-numeric channels: {non_numeric!r}\n"
                "Convert every channel column to real numeric values before reading."
            ) from exc

        _shape = result.shape
        logger.debug(f"CSV read complete, returning data with shape {_shape}")
        return result

Attributes

supported_extensions = ['.csv'] class-attribute

Functions

get_file_info(path, **kwargs) classmethod

Get basic information about the CSV file.

Parameters:

Name	Type	Description	Default
path	str | Path | bytes | bytearray | memoryview | BinaryIO	Union[str, Path]. Path to the CSV file.	required
**kwargs	Any	Any. Additional parameters for CSV reading. Supported parameters: delimiter : str, default="," Delimiter character. header : Optional[int], default=0 Row number to use as header. Set to None if no header. time_column : Union[int, str], default=0 Index or name of the time column.	{}

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Dictionary containing file information including: - samplerate: Estimated sampling rate in Hz - channels: Number of data channels (excluding time column) - frames: Total number of frames - format: "CSV" - duration: Duration in seconds (or None if cannot be calculated) - ch_labels: List of channel labels

Notes

This method accepts CSV-specific parameters through kwargs. See CSVFileInfoParams for supported parameter types.

Source code in wandas/io/readers.py

@classmethod
def get_file_info(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    **kwargs: Any,
) -> dict[str, Any]:
    """Get basic information about the CSV file.

    Args:
        path: Union[str, Path]. Path to the CSV file.
        **kwargs: Any. Additional parameters for CSV reading. Supported parameters:

            - delimiter : str, default=","
            Delimiter character.
            - header : Optional[int], default=0
            Row number to use as header. Set to None if no header.
            - time_column : Union[int, str], default=0
            Index or name of the time column.

    Returns:
        dict[str, Any]: Dictionary containing file information including:
            - samplerate: Estimated sampling rate in Hz
            - channels: Number of data channels (excluding time column)
            - frames: Total number of frames
            - format: "CSV"
            - duration: Duration in seconds (or None if cannot be calculated)
            - ch_labels: List of channel labels

    Notes:
        This method accepts CSV-specific parameters through kwargs.
        See CSVFileInfoParams for supported parameter types.
    """
    # Extract parameters with defaults
    delimiter: str = kwargs.get("delimiter", ",")
    header: int | None = kwargs.get("header", 0)
    time_column: int | str = kwargs.get("time_column", 0)

    # Parse the complete table because exact frame count and sampling rate are
    # required before the public Dask array can be constructed.
    pd = require_pandas("CSV file reading")
    df = pd.read_csv(_prepare_file_source(path), delimiter=delimiter, header=header)
    time_index = _resolve_csv_time_column(df.columns.tolist(), time_column)

    # Estimate sampling rate from the selected time column.
    grid_deviation = 0.0
    allowed_deviation = 0.0
    mean_interval = 0.0
    try:
        time_series = df.iloc[:, time_index]
        time_values = np.asarray(time_series.values, dtype=np.float64)
        time_start = float(time_values[0]) if len(time_values) > 0 else 0.0
        intervals = np.diff(time_values)
        if len(time_values) > 1 and np.all(np.isfinite(time_values)) and np.all(intervals > 0):
            mean_interval = float(np.mean(intervals))
            float_tolerance = 4 * float(np.spacing(np.max(np.abs(time_values))))
            rounding_allowance = _printed_rounding_allowance(time_values, intervals, float_tolerance)
            if rounding_allowance > 0:
                # Rounded times: accept exactly what rounding a uniform grid produces.
                grid_deviation = _uniform_grid_deviation(time_values)
                allowed_deviation = (
                    max(_CSV_MAX_GRID_DEVIATION * mean_interval, rounding_allowance)
                    + float_tolerance
                    + _GRID_SEARCH_TOLERANCE * mean_interval
                )
            else:
                # Exact times: anchor the grid at the first time, which keeps a
                # dropped sample detectable even in a file of a few rows.
                grid = time_start + mean_interval * np.arange(len(time_values))
                grid_deviation = float(np.max(np.abs(time_values - grid)))
                allowed_deviation = _CSV_MAX_GRID_DEVIATION * mean_interval + float_tolerance
            estimated_sr = float(1 / mean_interval)
            # Correct only integer-near roundoff; fractional rates are valid.
            nearest_integer = round(estimated_sr)
            if nearest_integer > 0 and np.isclose(estimated_sr, nearest_integer, rtol=1e-12, atol=0):
                estimated_sr = float(nearest_integer)
        else:
            estimated_sr = 0.0  # Cannot infer a valid rate from these coordinates
    except Exception:
        estimated_sr = 0.0  # Default if can't calculate
        time_start = 0.0

    if grid_deviation > allowed_deviation:
        raise ValueError(
            "Irregular CSV time column\n"
            f"  Got: a time value {grid_deviation / mean_interval:.3g} sample intervals away from a uniform grid\n"
            f"  Expected: at most {allowed_deviation / mean_interval:.3g} intervals (a quarter interval, or\n"
            "    half the printed time step when the times are rounded, plus float64 resolution)\n"
            "Wandas Frames assume a constant sampling rate, so irregular times would be\n"
            "relabelled silently. Resample the data to a uniform time grid before loading,\n"
            "or write the times with more decimals if they were rounded when saved."
        )

    channel_labels = [str(column) for index, column in enumerate(df.columns) if index != time_index]
    frames = df.shape[0]
    duration = frames / estimated_sr if estimated_sr else None

    # Return file info
    return {
        "samplerate": estimated_sr,
        "channels": len(channel_labels),
        "frames": frames,
        "format": "CSV",
        "duration": duration,
        "ch_labels": channel_labels,
        "time_start": time_start,
    }

get_data(path, channels, start_idx, frames, **kwargs) classmethod

Read data from the CSV file.

Parameters:

Name	Type	Description	Default
path	str | Path | bytes | bytearray | memoryview | BinaryIO	Union[str, Path]. Path to the CSV file.	required
channels	list[int]	list[int]. List of channel indices to read.	required
start_idx	int	int. Starting frame index.	required
frames	int	int. Number of frames to read.	required
**kwargs	Any	Any. Additional parameters for CSV reading. Supported parameters: delimiter : str, default="," Delimiter character. header : Optional[int], default=0 Row number to use as header. time_column : Union[int, str], default=0 Index or name of the time column.	{}

Returns:

Name	Type	Description
ArrayLike	ArrayLike	Array of shape (channels, frames) containing the data.

Notes

This method accepts CSV-specific parameters through kwargs. See CSVGetDataParams for supported parameter types.

Source code in wandas/io/readers.py

@classmethod
def get_data(
    cls,
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    channels: list[int],
    start_idx: int,
    frames: int,
    **kwargs: Any,
) -> ArrayLike:
    """Read data from the CSV file.

    Args:
        path: Union[str, Path]. Path to the CSV file.
        channels: list[int]. List of channel indices to read.
        start_idx: int. Starting frame index.
        frames: int. Number of frames to read.
        **kwargs: Any. Additional parameters for CSV reading. Supported parameters:

            - delimiter : str, default=","
            Delimiter character.
            - header : Optional[int], default=0
            Row number to use as header.
            - time_column : Union[int, str], default=0
            Index or name of the time column.

    Returns:
        ArrayLike: Array of shape (channels, frames) containing the data.

    Notes:
        This method accepts CSV-specific parameters through kwargs.
        See CSVGetDataParams for supported parameter types.
    """
    # Extract parameters with defaults
    time_column: int | str = kwargs.get("time_column", 0)
    delimiter: str = kwargs.get("delimiter", ",")
    header: int | None = kwargs.get("header", 0)

    logger.debug(f"Reading CSV data from {path!r} starting at {start_idx}")

    # Read the CSV file
    pd = require_pandas("CSV file reading")
    df = pd.read_csv(_prepare_file_source(path), delimiter=delimiter, header=header)
    time_index = _resolve_csv_time_column(df.columns.tolist(), time_column)

    # Remove time column
    df = df.iloc[:, [index for index in range(df.shape[1]) if index != time_index]]

    # Select requested channels - adjust indices to account for time column removal
    if channels:
        try:
            data_df = df.iloc[:, channels]
        except IndexError as e:
            raise ValueError(f"Requested channels {channels} out of range") from e
    else:
        data_df = df

    # Handle start_idx and frames for partial reading
    end_idx = start_idx + frames if frames > 0 else None
    data_df = data_df.iloc[start_idx:end_idx]

    # Convert to numpy array and transpose to (channels, samples) format
    try:
        result = data_df.to_numpy(dtype=np.float64).T
    except (AttributeError, TypeError, ValueError) as exc:
        non_numeric = [
            str(column) for column in data_df.columns if not pd.api.types.is_numeric_dtype(data_df[column])
        ]
        if not non_numeric:
            raise ValueError("Unexpected data type after reading file") from exc
        raise ValueError(
            "CSV data channels must be numeric\n"
            f"  Non-numeric channels: {non_numeric!r}\n"
            "Convert every channel column to real numeric values before reading."
        ) from exc

    _shape = result.shape
    logger.debug(f"CSV read complete, returning data with shape {_shape}")
    return result

Functions

download_url_to_temporary_file(url, *, timeout, suffix=None, resource_name='file', max_bytes=None, chunk_size=None)

Source code in wandas/io/readers.py

def download_url_to_temporary_file(
    url: str,
    *,
    timeout: float,
    suffix: str | None = None,
    resource_name: str = "file",
    max_bytes: int | None = None,
    chunk_size: int | None = None,
) -> DownloadedTemporaryFile:
    import urllib.error
    import urllib.request

    effective_max_bytes = MAX_URL_DOWNLOAD_BYTES if max_bytes is None else max_bytes
    effective_chunk_size = URL_DOWNLOAD_CHUNK_SIZE if chunk_size is None else chunk_size
    if effective_max_bytes <= 0:
        raise ValueError(
            f"Download size limit must be greater than zero\n"
            f"  Resource: {resource_name}\n"
            f"  URL: {url}\n"
            f"  Got: {effective_max_bytes} bytes\n"
            f"Provide a positive max_bytes value."
        )
    if effective_chunk_size <= 0:
        raise ValueError(
            f"Download chunk size must be greater than zero\n"
            f"  Resource: {resource_name}\n"
            f"  URL: {url}\n"
            f"  Got: {effective_chunk_size} bytes\n"
            f"Provide a positive chunk_size value."
        )
    normalized_suffix = _normalize_extension(suffix) or ""
    downloaded_bytes = 0
    temp_dir: tempfile.TemporaryDirectory[str] | None = None
    downloaded_file: DownloadedTemporaryFile | None = None

    try:
        with urllib.request.urlopen(url, timeout=timeout) as response:
            content_length = _get_validated_content_length_or_none(
                response,
                url=url,
                resource_name=resource_name,
            )
            if content_length is not None and content_length > effective_max_bytes:
                raise OSError(
                    f"Declared size of {resource_name} exceeds download limit\n"
                    f"  URL: {url}\n"
                    f"  Declared size: {content_length} bytes\n"
                    f"  Limit: {effective_max_bytes} bytes\n"
                    f"Use a smaller file or download it locally before loading."
                )

            temp_dir = tempfile.TemporaryDirectory()
            temp_path = Path(temp_dir.name) / f"download{normalized_suffix}"
            downloaded_file = DownloadedTemporaryFile(path=temp_path, temp_dir=temp_dir)
            with temp_path.open("wb") as temp_file:
                while True:
                    remaining_bytes = effective_max_bytes - downloaded_bytes
                    read_size = min(effective_chunk_size, remaining_bytes + 1)
                    chunk = response.read(read_size)
                    if not chunk:
                        break
                    next_downloaded_bytes = downloaded_bytes + len(chunk)
                    if next_downloaded_bytes > effective_max_bytes:
                        raise OSError(
                            f"Streaming {resource_name} would exceed size limit\n"
                            f"  URL: {url}\n"
                            f"  Attempted size: {next_downloaded_bytes} bytes\n"
                            f"  Limit: {effective_max_bytes} bytes\n"
                            f"Use a smaller file or download it locally before loading."
                        )
                    downloaded_bytes = next_downloaded_bytes
                    temp_file.write(chunk)
            return downloaded_file
    except urllib.error.URLError as exc:
        if downloaded_file is not None:
            downloaded_file.cleanup()
        elif temp_dir is not None:
            temp_dir.cleanup()
        raise OSError(
            f"Failed to download {resource_name} from URL\n"
            f"  URL: {url}\n"
            f"  Error: {exc}\n"
            f"Verify the URL is accessible and try again."
        ) from exc
    except Exception:
        if downloaded_file is not None:
            downloaded_file.cleanup()
        elif temp_dir is not None:
            temp_dir.cleanup()
        raise

supported_formats()

Return file extensions supported by the registered readers.

Source code in wandas/io/readers.py

def supported_formats() -> list[str]:
    """Return file extensions supported by the registered readers."""
    extensions: set[str] = set()
    for reader in _file_readers:
        extensions.update(reader.__class__._normalize_supported_extensions())
    return sorted(extensions)

get_file_reader(path, *, file_type=None)

Return the registered reader selected by an explicit type or path suffix.

file_type is case-insensitive and may include or omit the leading dot. This lower-level registry boundary does not guess a format from file contents. The public :func:wandas.read entry point owns the additional name-based inference and anonymous-WAV compatibility default.

Raises:

Type	Description
ValueError	If neither a type nor suffix is available, or if the normalized extension has no registered reader.

Source code in wandas/io/readers.py

def get_file_reader(
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    *,
    file_type: str | None = None,
) -> FileReader:
    """Return the registered reader selected by an explicit type or path suffix.

    ``file_type`` is case-insensitive and may include or omit the leading dot.
    This lower-level registry boundary does not guess a format from file
    contents. The public :func:`wandas.read` entry point owns the additional
    name-based inference and anonymous-WAV compatibility default.

    Raises:
        ValueError: If neither a type nor suffix is available, or if the
            normalized extension has no registered reader.
    """
    path_str = str(path)
    ext = _normalize_extension(file_type)
    if ext is None and isinstance(path, (str, Path)):
        ext = Path(path).suffix.lower()
    if not ext:
        raise ValueError(
            "File type is required when the extension is missing\n"
            "  Cannot determine format without an extension\n"
            "  Provide file_type like '.wav' or '.csv'"
        )

    # Try each reader in order
    for reader in _file_readers:
        if ext in reader.__class__._normalize_supported_extensions():
            logger.debug(f"Using {reader.__class__.__name__} for {path_str}")
            return reader

    # If no reader found, raise error
    raise ValueError(f"No suitable file reader found for {path_str}")

register_file_reader(reader_class)

Register a new file reader.

Source code in wandas/io/readers.py

def register_file_reader(reader_class: type) -> None:
    """Register a new file reader."""
    reader = reader_class()
    _file_readers.append(reader)
    logger.debug(f"Registered new file reader: {reader_class.__name__}")

wandas.io.wav_io

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions

write_wav(filename, target, format=None)

Write a ChannelFrame object to a WAV file.

Floating-point samples use IEEE FLOAT regardless of amplitude, preserving values outside [-1, 1] without normalization or PCM clipping. Values are stored at float32 precision. Integer input retains the format's default subtype. Writing computes the calibrated samples synchronously.

Parameters:

Name	Type	Description	Default
filename	str | PathLike[str] | BinaryIO	Path or writable binary stream. Caller-owned streams remain open. Stream output is encoded in memory and written once at the current position, so non-seekable streams and streams with existing content are supported.	required
target	ChannelFrame	ChannelFrame. ChannelFrame object containing the data to write.	required
format	str | None	Required for streams. Paths infer the format from their extension.	None

Raises:

Type	Description
ValueError	If target is not a ChannelFrame or a stream has no format.

Source code in wandas/io/wav_io.py

def write_wav(filename: str | os.PathLike[str] | BinaryIO, target: "ChannelFrame", format: str | None = None) -> None:
    """
    Write a ChannelFrame object to a WAV file.

    Floating-point samples use IEEE FLOAT regardless of amplitude, preserving
    values outside [-1, 1] without normalization or PCM clipping. Values are
    stored at float32 precision. Integer input retains the format's default
    subtype. Writing computes the calibrated samples synchronously.

    Args:
        filename: Path or writable binary stream. Caller-owned streams remain open.
            Stream output is encoded in memory and written once at the current
            position, so non-seekable streams and streams with existing content
            are supported.
        target: ChannelFrame. ChannelFrame object containing the data to write.
        format: Required for streams. Paths infer the format from their extension.

    Raises:
        ValueError: If target is not a ChannelFrame or a stream has no format.
    """
    from wandas.frames.channel import ChannelFrame

    if not isinstance(target, ChannelFrame):
        raise ValueError("target must be a ChannelFrame object.")

    stream: BinaryIO | None = None
    buffer: io.BytesIO | None = None
    if isinstance(filename, (str, os.PathLike)):
        sink: str | io.BytesIO = os.fspath(filename)
    else:
        if format is None:
            raise ValueError("format is required when writing to a binary stream")
        # libsndfile seeks back to patch the header, which fails silently on
        # non-seekable streams and assumes the stream starts at offset 0. Encode
        # into a private buffer and hand the caller's stream finished bytes.
        stream = filename
        sink = buffer = io.BytesIO()

    logger.debug(f"Saving audio data to file: {filename} (will compute now)")
    data = target._compute()
    data = data.T
    if data.shape[1] == 1:
        data = data.squeeze(axis=1)
    if np.issubdtype(data.dtype, np.floating):
        sf.write(
            sink,
            data,
            int(target.sampling_rate),
            subtype="FLOAT",
            format=format,
        )
    else:
        sf.write(sink, data, int(target.sampling_rate), format=format)
    if stream is not None and buffer is not None:
        _write_all(stream, buffer.getbuffer())
    logger.debug(f"Save complete: {filename}")

wandas.io.wdf_io

Strict xarray-backed persistence for typed WDF 0.4/0.5/0.6 artifacts.

Attributes

WDF_FORMAT_VERSION = '0.4' module-attribute

__all__ = ['WDF_FORMAT_VERSION', 'load', 'save'] module-attribute

Classes

Functions

save(frame, path, *, compress='gzip', overwrite=False)

Save an exact built-in Frame as WDF 0.4, 0.5 or sparse 0.6.

A time-frequency Frame with a known physical frame origin uses WDF 0.5. Sparse time-frequency results use WDF 0.6 with explicit sparse state. All other Frames retain WDF 0.4. Older readers reject new versions explicitly; this preserves placement instead of silently discarding its additional state.

The artifact retains the Frame label, user metadata, channel labels and metadata (units, references, calibration, and channel extras), source-time offsets, analysis coordinates, and the derived operation_history view. Metadata fields use strict JSON encoding and therefore follow JSON value semantics (for example, a tuple is loaded as a list). It stores the concrete Frame result; previous references and replayable Recipe intent are not persisted.

Dask data is handed directly to xarray and is written synchronously; Wandas does not first materialize the complete tensor with frame._data.compute(). The frame is written to a temporary file first, so a failure while computing or writing it leaves an existing file unchanged and creates no partial file. A new file is then moved into place. An existing file is overwritten in place by copying the finished temporary file into it, so it keeps its hard links, owner, group, permission bits and ACLs, and a symlinked path updates the file it points to. Overwriting a file that is not writable, or an HDF5 file that is open or locked, raises OSError before anything is computed. An interruption of that final copy (for example a full disk or a power loss), or another process opening the file between that check and the copy, can still leave the existing file incomplete.

Saving needs the io extra (pip install "wandas[io]"); without it, ImportError names that install command.

Source code in wandas/io/wdf_io.py

def save(
    frame: BaseFrame[Any],
    path: str | Path,
    *,
    compress: str | None = "gzip",
    overwrite: bool = False,
) -> None:
    """Save an exact built-in Frame as WDF 0.4, 0.5 or sparse 0.6.

    A time-frequency Frame with a known physical frame origin uses WDF 0.5.
    Sparse time-frequency results use WDF 0.6 with explicit sparse state.
    All other Frames retain WDF 0.4. Older readers reject new versions explicitly; this
    preserves placement instead of silently discarding its additional state.

    The artifact retains the Frame label, user metadata, channel labels and
    metadata (units, references, calibration, and channel extras), source-time
    offsets, analysis coordinates, and the derived ``operation_history`` view.
    Metadata fields use strict JSON encoding and therefore follow JSON value
    semantics (for example, a tuple is loaded as a list).
    It stores the concrete Frame result; ``previous`` references and replayable
    Recipe intent are not persisted.

    Dask data is handed directly to xarray and is written synchronously; Wandas does
    not first materialize the complete tensor with ``frame._data.compute()``.
    The frame is written to a temporary file first, so a failure while computing
    or writing it leaves an existing file unchanged and creates no partial file.
    A new file is then moved into place. An existing file is overwritten in place
    by copying the finished temporary file into it, so it keeps its hard links,
    owner, group, permission bits and ACLs, and a symlinked path updates the file
    it points to. Overwriting a file that is not writable, or an HDF5 file that is
    open or locked, raises ``OSError`` before anything is computed. An
    interruption of that final copy (for example a full disk or a power loss), or
    another process opening the file between that check and the copy, can still
    leave the existing file incomplete.

    Saving needs the ``io`` extra (``pip install "wandas[io]"``); without it,
    ``ImportError`` names that install command.
    """
    target = _normalized_path(path)
    if target.exists() and not overwrite:
        raise FileExistsError(f"File {target} already exists. Set overwrite=True to overwrite.")

    # Validate all Frame and JSON state before either importing the storage backend
    # or opening the destination, so invalid state cannot leave a partial artifact.
    dataset = _build_dataset(frame)
    require_h5netcdf("WDF save")
    encoding = (
        {"data": {"compression": compress, "chunksizes": _data_chunksizes(dataset["data"])}} if compress else None
    )
    # Lazy data is computed while writing, so the frame is written to a temporary
    # file first: a failure leaves an existing artifact intact and no partial file.
    destination = Path(os.path.realpath(target))
    if not destination.exists():
        # A fixed-length name stays within NAME_MAX even for a maximal destination name.
        temporary = destination.with_name(f".wdf-{uuid.uuid4().hex}.tmp")
        os.close(os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o666))
        try:
            _write_dataset(dataset, temporary, encoding)
            os.replace(temporary, destination)
        except BaseException:
            temporary.unlink(missing_ok=True)
            raise
        return

    # Copy the finished file into the existing one so it keeps its inode: hard
    # links, owner, group, permission bits and ACLs stay as they were.
    # Opening it for writing without truncation reports a read-only destination
    # before the frame is computed.
    os.close(os.open(destination, os.O_WRONLY))
    _require_hdf5_not_open(destination)
    try:
        handle, name = tempfile.mkstemp(prefix=".wdf-", suffix=".tmp", dir=destination.parent)
    except PermissionError:
        handle, name = tempfile.mkstemp(prefix="wdf-", suffix=".tmp")
    os.close(handle)
    temporary = Path(name)
    try:
        _write_dataset(dataset, temporary, encoding)
        shutil.copyfile(temporary, destination)
    finally:
        temporary.unlink(missing_ok=True)

load(path)

Load a local WDF 0.4, 0.5 or 0.6 artifact as its exact built-in Frame type.

WDF 0.4 time-frequency Frames have unknown physical center times; their existing zero-based time axes remain available. WDF 0.5 restores the explicit physical origin without inferring it from history. WDF 0.6 also restores the sparse-window permission, retaining gaps in the time axis.

The returned Frame restores the saved label, user metadata, channel labels and metadata (units, references, calibration, and channel extras), source-time offsets, analysis coordinates, and derived operation_history view. Metadata follows strict JSON value semantics; for example, a tuple saved in metadata is loaded as a list. WDF does not restore a previous reference or replayable Recipe intent.

The returned Frame owns access to its source internally. Keep the source path unchanged while that Frame or Frames derived from it are in use. Obtain NumPy values through frame.data without managing the storage backend directly.

Loading needs the io extra (pip install "wandas[io]"); without it, ImportError names that install command.

Source code in wandas/io/wdf_io.py

def load(path: str | Path) -> BaseFrame[Any]:
    """Load a local WDF 0.4, 0.5 or 0.6 artifact as its exact built-in Frame type.

    WDF 0.4 time-frequency Frames have unknown physical center times; their
    existing zero-based time axes remain available. WDF 0.5 restores the
    explicit physical origin without inferring it from history. WDF 0.6 also
    restores the sparse-window permission, retaining gaps in the time axis.

    The returned Frame restores the saved label, user metadata, channel labels
    and metadata (units, references, calibration, and channel extras),
    source-time offsets, analysis coordinates, and derived
    ``operation_history`` view. Metadata follows strict JSON value semantics;
    for example, a tuple saved in metadata is loaded as a list. WDF does not
    restore a ``previous`` reference or replayable Recipe intent.

    The returned Frame owns access to its source internally. Keep the source path
    unchanged while that Frame or Frames derived from it are in use. Obtain NumPy
    values through ``frame.data`` without managing the storage backend directly.

    Loading needs the ``io`` extra (``pip install "wandas[io]"``); without it,
    ``ImportError`` names that install command.
    """
    source = Path(path)
    if not source.exists():
        raise FileNotFoundError(f"File not found: {source}")
    require_h5netcdf("WDF load")

    # CF decoding is disabled because WDF owns dtype/value semantics; foreign CF
    # scale, offset, fill-value, and time attributes must never transform raw data.
    dataset = xr.open_dataset(
        source,
        engine="h5netcdf",
        chunks={},
        decode_cf=False,
        mask_and_scale=False,
        backend_kwargs={"phony_dims": "access"},
    )
    version = dataset.attrs.get("version")
    if version not in (WDF_FORMAT_VERSION, _FRAME_TIME_FORMAT_VERSION, _SPARSE_FORMAT_VERSION):
        got = "missing" if version is None else repr(version)
        raise ValueError(
            "Unsupported WDF format version\n"
            f"  Got: {got}\n"
            f"  Supported: {WDF_FORMAT_VERSION!r}, {_FRAME_TIME_FORMAT_VERSION!r}, {_SPARSE_FORMAT_VERSION!r}\n"
            "Use a compatible Wandas version or resave the file."
        )
    _require_exact_schema(dataset)

    frame_type = dataset.attrs["frame_type"]
    if not isinstance(frame_type, str):
        raise ValueError("Invalid WDF frame_type; expected text")
    constructor = _load_json(dataset.attrs["constructor_json"], field="constructor_json")
    has_origin = isinstance(constructor, dict) and "frame_time_origin" in constructor
    has_sparse = isinstance(constructor, dict) and "allow_sparse" in constructor
    if version == _SPARSE_FORMAT_VERSION:
        if frame_type not in {"SpectrogramFrame", "CepstrogramFrame"} or not has_sparse:
            raise ValueError("Invalid WDF 0.6 sparse schema; expected sparse time-frequency state")
    elif has_sparse:
        raise ValueError("WDF sparse state requires format version 0.6")
    if version == _FRAME_TIME_FORMAT_VERSION:
        if frame_type not in {"SpectrogramFrame", "CepstrogramFrame"} or not has_origin:
            raise ValueError("Invalid WDF 0.5 physical frame-time schema; expected an explicit time-frequency origin")
    elif has_origin and version != _SPARSE_FORMAT_VERSION:
        raise ValueError("WDF physical frame-time state requires format version 0.5")
    metadata = _load_json(dataset.attrs["metadata_json"], field="metadata_json")
    label = _load_json(dataset.attrs["label"], field="label")
    history = _validate_history(_load_json(dataset.attrs["operation_history_json"], field="operation_history_json"))
    if not isinstance(constructor, Mapping):
        raise ValueError("Invalid WDF constructor_json; expected an object")
    if not isinstance(metadata, dict):
        raise ValueError("Invalid WDF metadata_json; expected an object")
    if label is not None and not isinstance(label, str):
        raise ValueError("Invalid WDF label; expected a string or null")
    sampling_rate = _finite_number(dataset.attrs["sampling_rate"], field="sampling_rate")

    data = dataset["data"]
    channel_count = int(dataset.sizes["channel"])
    channel_ids = _text_vector(dataset, "channel", channel_count)
    labels = _text_vector(dataset, "channel_label", channel_count)
    units = _text_vector(dataset, "channel_unit", channel_count)
    refs = _number_vector(dataset, "channel_ref", channel_count)
    factors = _number_vector(dataset, "channel_calibration_factor", channel_count)
    offsets = _number_vector(dataset, "source_time_offset", channel_count)
    extras_json = _text_vector(dataset, "channel_extra_json", channel_count)
    extras = [_load_json(value, field=f"channel_extra_json[{index}]") for index, value in enumerate(extras_json)]
    if not all(isinstance(extra, dict) for extra in extras):
        raise ValueError("Invalid WDF channel_extra_json; expected JSON objects")
    channels = [
        ChannelMetadata(
            label=labels[index],
            calibration=ChannelCalibration(factor=float(factors[index]), unit=units[index], ref=float(refs[index])),
            extra=extras[index],
        )
        for index in range(channel_count)
    ]
    common: dict[str, Any] = {
        "sampling_rate": sampling_rate,
        "label": label,
        "metadata": metadata,
        "channel_metadata": channels,
        "channel_ids": channel_ids,
        "source_time_offset": offsets,
        "operation_history_prefix": history,
    }
    frame = decode_frame(
        frame_type,
        constructor,
        data=data.data,
        common=common,
        stored_dims=cast(tuple[str, ...], tuple(data.dims)),
    )
    coordinates = {
        str(name): np.asarray(coordinate.values) for name, coordinate in dataset.coords.items() if name != "channel"
    }
    restore_frame_coordinates(frame, coordinates)
    return frame