IO Module / 入出力モジュール¶

The wandas.io module provides reading and writing capabilities for various file formats. wandas.io モジュールは、様々なファイル形式の読み書き機能を提供します。

Recommended Entry Points / 推奨入口¶

Use wd.read(...) for external source data such as WAV, CSV, supported audio files, URLs, bytes, and file-like objects. WAV、CSV、対応音声ファイル、URL、bytes、file-like object などの外部ソースデータには wd.read(...) を使います。

Use wd.load(...) for Wandas native WDF files. Wandas native WDF ファイルには wd.load(...) を使います。

read_wav() and read_csv() remain available for compatibility, but new documentation and examples prefer read(). 互換性のため read_wav() と read_csv() は残りますが、新しいドキュメントと例では read() を優先します。

Canonical numeric contract / 正規化数値契約¶

Built-in readers always produce lazy, channel-first float64 data. Equal file content has equal values whether it comes from a local path, URL, bytes, bytearray, memoryview, or a file-like object.

built-in readerは常に遅延実行のchannel-first float64を返します。同じファイル内容なら、 local path、URL、bytes、bytearray、memoryview、file-like objectのどれから読んでも値は同じです。

Input / 入力	`wd.read()` numeric rule / 数値規則
WAV (`PCM_U8`, `PCM_16`, `PCM_24`, `PCM_32`)	libsndfile full-scale conversion; unsigned 8-bit PCM is zero-centered / libsndfileのfull-scale変換、符号なし8-bit PCMもゼロ中心
WAV (`FLOAT`, `DOUBLE`)	Values are preserved as `float64`; no clipping, so values may exceed ±1 / 値を`float64`で保持し、クリップしないため±1を超える場合がある
FLAC, OGG, AIFF/AIF, SND	libsndfile full-scale `float64` audio / libsndfileのfull-scale `float64`音声
CSV	Non-time numeric values are preserved and cast to `float64`; non-numeric channels are rejected / 時間列以外の数値を維持して`float64`化し、非数値chは拒否

This is decode normalization, not peak normalization: Wandas never divides by the maximum value of an individual waveform. frame.normalize() and playback's normalize option remain separate processing and presentation features.

これは波形ごとの最大値で割るpeak normalizationではなく、decode時の正規化です。 frame.normalize()と再生時のnormalizeは別の処理・表示機能です。

Migration / 移行¶

Local integer WAV files previously defaulted to raw PCM counts cast to float32. They now use the same full-scale float64 decoding as every other transport. Calibration factors derived for raw counts must be derived again from a reference recording read under the new contract. wd.load() preserves the dtype stored in WDF, and wd.from_numpy() preserves the user-selected array dtype; neither contract changes here.

従来local integer WAVはraw PCM countをfloat32へcastしていましたが、今後は他の transportと同じfull-scale float64です。raw count向けの既知係数は、新契約で読んだ参照収録から再導出してください。wd.load()はWDF保存dtype、wd.from_numpy()は利用者指定dtypeを維持し、これらの契約は変更しません。

File Readers / ファイルリーダー¶

Provides functionality to read data from various file formats. 様々なファイル形式からデータを読み込む機能を提供します。

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

Parameters¶

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

Source code in wandas/io/readers.py

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

    Parameters
    ----------
    delimiter : str
        Delimiter character. Default is ",".
    header : Optional[int]
        Row number to use as header. Default is 0 (first row).
        Set to None if no header.
    time_column : Union[int, str]
        Index or name of the time column. Default is 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.

Parameters¶

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

Source code in wandas/io/readers.py

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

    Parameters
    ----------
    delimiter : str
        Delimiter character. Default is ",".
    header : Optional[int]
        Row number to use as header. Default is 0.
    time_column : Union[int, str]
        Index or name of the time column. Default is 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:
`dict[str, Any]`	samplerate: Sampling rate in Hz
`dict[str, Any]`	channels: Number of channels
`dict[str, Any]`	frames: Total number of frames
`dict[str, Any]`	format: File format
`dict[str, Any]`	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."""

    # SoundFile supported formats
    supported_extensions: ClassVar[list[str]] = [".wav", ".flac", ".ogg", ".aiff", ".aif", ".snd"]

    @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,
        }

    @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 float64 audio data from the file."""
        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)
            data = f.read(frames=frames, dtype="float64", always_2d=True)

            # Select requested channels
            data = data[:, channels]

            # Transpose to get (channels, samples) format
            result = data.T
            if not isinstance(result, np.ndarray):
                raise ValueError("Unexpected data type after reading file")

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

Attributes¶

`supported_extensions = ['.wav', '.flac', '.ogg', '.aiff', '.aif', '.snd']` `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,
    }

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

Read full-scale float64 audio data from the file.

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 float64 audio data from the file."""
    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)
        data = f.read(frames=frames, dtype="float64", always_2d=True)

        # Select requested channels
        data = data[:, channels]

        # Transpose to get (channels, samples) format
        result = data.T
        if not isinstance(result, np.ndarray):
            raise ValueError("Unexpected data type after reading file")

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

Source code in wandas/io/readers.py

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

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

        Parameters
        ----------
        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)

        # Read first few lines to determine structure
        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.
        try:
            time_series = df.iloc[:, time_index]
            time_values = np.array(time_series.values)
            time_start = float(time_values[0]) if len(time_values) > 0 else 0.0
            if len(time_values) > 1:
                # Use round() instead of int() to handle floating-point precision issues
                estimated_sr = round(1 / np.mean(np.diff(time_values)))
            else:
                estimated_sr = 0  # Cannot determine from single row
        except Exception:
            estimated_sr = 0  # Default if can't calculate
            time_start = 0.0

        channel_labels = [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.

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

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.

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.

    Parameters
    ----------
    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)

    # Read first few lines to determine structure
    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.
    try:
        time_series = df.iloc[:, time_index]
        time_values = np.array(time_series.values)
        time_start = float(time_values[0]) if len(time_values) > 0 else 0.0
        if len(time_values) > 1:
            # Use round() instead of int() to handle floating-point precision issues
            estimated_sr = round(1 / np.mean(np.diff(time_values)))
        else:
            estimated_sr = 0  # Cannot determine from single row
    except Exception:
        estimated_sr = 0  # Default if can't calculate
        time_start = 0.0

    channel_labels = [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¶

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.

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.

    Parameters
    ----------
    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)` ¶

Get an appropriate file reader for the given path or file type.

Source code in wandas/io/readers.py

def get_file_reader(
    path: str | Path | bytes | bytearray | memoryview | BinaryIO,
    *,
    file_type: str | None = None,
) -> FileReader:
    """Get an appropriate file reader for the given path or file type."""
    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__}")

WAV File IO / WAVファイル入出力¶

Provides functions for reading and writing WAV files. WAVファイルの読み書き機能を提供します。

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

Parameters¶

filename : str Path to the WAV file. target : ChannelFrame ChannelFrame object containing the data to write. format : str, optional File format. If None, determined from file extension.

Raises¶

ValueError If target is not a ChannelFrame object.

Source code in wandas/io/wav_io.py

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

    Parameters
    ----------
    filename : str
        Path to the WAV file.
    target : ChannelFrame
        ChannelFrame object containing the data to write.
    format : str, optional
        File format. If None, determined from file extension.

    Raises
    ------
    ValueError
        If target is not a ChannelFrame object.
    """
    from wandas.frames.channel import ChannelFrame

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

    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) and np.max(np.abs(data)) <= 1:
        sf.write(
            str(filename),
            data,
            int(target.sampling_rate),
            subtype="FLOAT",
            format=format,
        )
    else:
        sf.write(str(filename), data, int(target.sampling_rate), format=format)
    logger.debug(f"Save complete: {filename}")

WDF File IO / WDFファイル入出力¶

Provides functions for reading and writing WDF (Wandas Data File) format, which enables complete preservation including metadata. WDF（Wandas Data File）形式の読み書き機能を提供します。このフォーマットはメタデータを含む完全な保存が可能です。

`wandas.io.wdf_io` ¶

Strict xarray-backed persistence for typed WDF 0.4 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.

Dask data is handed directly to xarray and is written synchronously; Wandas does not first materialize the complete tensor with frame._data.compute().

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.

    Dask data is handed directly to xarray and is written synchronously; Wandas does
    not first materialize the complete tensor with ``frame._data.compute()``.
    """
    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}} if compress else None
    dataset.to_netcdf(
        target,
        engine="h5netcdf",
        encoding=encoding,
        invalid_netcdf=True,
    )

`load(path)` ¶

Load a local WDF 0.4 artifact as its exact built-in Frame type.

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.

Source code in wandas/io/wdf_io.py

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

    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.
    """
    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 != WDF_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}\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")
    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

IO Module / 入出力モジュール¶

Recommended Entry Points / 推奨入口¶

Canonical numeric contract / 正規化数値契約¶

Migration / 移行¶

File Readers / ファイルリーダー¶

wandas.io.readers ¶

Attributes¶

logger = logging.getLogger(__name__) module-attribute ¶

Classes¶

DownloadedTemporaryFile dataclass ¶

Attributes¶

path instance-attribute ¶

temp_dir instance-attribute ¶

Functions¶

__init__(path, temp_dir) ¶

__post_init__() ¶

__enter__() ¶

__exit__(exc_type, exc, tb) ¶

cleanup() ¶

CSVFileInfoParams ¶

Parameters¶

Attributes¶

delimiter instance-attribute ¶

header instance-attribute ¶

time_column instance-attribute ¶

CSVGetDataParams ¶

Parameters¶

Attributes¶

delimiter instance-attribute ¶

header instance-attribute ¶

time_column instance-attribute ¶

FileReader ¶

Attributes¶

supported_extensions = [] class-attribute ¶

Functions¶

get_file_info(path, **kwargs) abstractmethod classmethod ¶

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

can_read(path) classmethod ¶

SoundFileReader ¶

Attributes¶

supported_extensions = ['.wav', '.flac', '.ogg', '.aiff', '.aif', '.snd'] class-attribute ¶

Functions¶

get_file_info(path, **kwargs) classmethod ¶

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

CSVFileReader ¶

Attributes¶

supported_extensions = ['.csv'] class-attribute ¶

Functions¶

get_file_info(path, **kwargs) classmethod ¶

Parameters¶

Returns¶

Notes¶

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

Parameters¶

Returns¶

Notes¶

Functions¶

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

supported_formats() ¶

get_file_reader(path, *, file_type=None) ¶

register_file_reader(reader_class) ¶

WAV File IO / WAVファイル入出力¶

wandas.io.wav_io ¶

Attributes¶

logger = logging.getLogger(__name__) module-attribute ¶

Classes¶

Functions¶

write_wav(filename, target, format=None) ¶

Parameters¶

Raises¶

WDF File IO / WDFファイル入出力¶

wandas.io.wdf_io ¶

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

load(path) ¶

`wandas.io.readers` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`DownloadedTemporaryFile` `dataclass` ¶

`path` `instance-attribute` ¶

`temp_dir` `instance-attribute` ¶

`init(path, temp_dir)` ¶

`__post_init__()` ¶

`enter()` ¶

`exit(exc_type, exc, tb)` ¶

`cleanup()` ¶

`CSVFileInfoParams` ¶

`delimiter` `instance-attribute` ¶

`header` `instance-attribute` ¶

`time_column` `instance-attribute` ¶

`CSVGetDataParams` ¶

`delimiter` `instance-attribute` ¶

`header` `instance-attribute` ¶

`time_column` `instance-attribute` ¶

`FileReader` ¶

`supported_extensions = []` `class-attribute` ¶

`get_file_info(path, **kwargs)` `abstractmethod` `classmethod` ¶

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

`can_read(path)` `classmethod` ¶

`SoundFileReader` ¶

`supported_extensions = ['.wav', '.flac', '.ogg', '.aiff', '.aif', '.snd']` `class-attribute` ¶

`get_file_info(path, **kwargs)` `classmethod` ¶

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

`CSVFileReader` ¶

`supported_extensions = ['.csv']` `class-attribute` ¶

`get_file_info(path, **kwargs)` `classmethod` ¶

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

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

`supported_formats()` ¶

`get_file_reader(path, *, file_type=None)` ¶

`register_file_reader(reader_class)` ¶

`wandas.io.wav_io` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`write_wav(filename, target, format=None)` ¶

`wandas.io.wdf_io` ¶

`WDF_FORMAT_VERSION = '0.4'` `module-attribute` ¶

`all = ['WDF_FORMAT_VERSION', 'load', 'save']` `module-attribute` ¶

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

`load(path)` ¶