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

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

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.

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

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

Parameters:

Name	Type	Description	Default
filename	str	str. Path to the WAV file.	required
target	ChannelFrame	ChannelFrame. ChannelFrame object containing the data to write.	required
format	str | None	str, optional. File format. If None, determined from file extension.	None

Raises:

Type	Description
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.

    Args:
        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}")

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.

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().

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.

    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()``.
    """
    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 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.

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