Utilities Module / ユーティリティモジュール

The wandas.utils module contains dataset, sample-generation, type, and helper APIs used by Wandas. Parameters, return values, exceptions, and examples are maintained in the generated docstrings.

wandas.utilsはdataset、sample生成、型、helper APIを提供します。引数、戻り値、例外、 使用例は生成されたdocstringで管理します。

wandas.utils.frame_dataset

Attributes

logger = logging.getLogger(__name__) module-attribute

FrameType = ChannelFrame | SpectrogramFrame module-attribute

F = TypeVar('F', bound=FrameType) module-attribute

F_out = TypeVar('F_out', bound=FrameType) module-attribute

MetadataResolver = Callable[[Path], Mapping[str, object]] module-attribute

Classes

LazyFrame dataclass

Bases: Generic[F]

A class that encapsulates a frame and its loading state.

Attributes:

Name	Type	Description
file_path	Path	File path associated with the frame
frame	F | None	Loaded frame object (None if not loaded)
is_loaded	bool	Flag indicating if the frame is loaded
load_attempted	bool	Flag indicating if loading was attempted (for error detection)

Source code in wandas/utils/frame_dataset.py

@dataclass
class LazyFrame(Generic[F]):
    """
    A class that encapsulates a frame and its loading state.

    Attributes:
        file_path: File path associated with the frame
        frame: Loaded frame object (None if not loaded)
        is_loaded: Flag indicating if the frame is loaded
        load_attempted: Flag indicating if loading was attempted (for error detection)
    """

    file_path: Path
    metadata: dict[str, object] = field(default_factory=dict)
    frame: F | None = None
    is_loaded: bool = False
    load_attempted: bool = False

    def ensure_loaded(self, loader: Callable[[Path], F | None]) -> F | None:
        """
        Ensures the frame is loaded, loading it if necessary.

        Args:
            loader: Function to load a frame from a file path

        Returns:
            The loaded frame, or None if loading failed
        """
        # Return the current frame if already loaded
        if self.is_loaded:
            return self.frame

        # Attempt to load if not loaded yet
        try:
            self.load_attempted = True
            self.frame = loader(self.file_path)
            self.is_loaded = True
            return self.frame
        except Exception as e:
            logger.error(f"Failed to load file {self.file_path}: {e!s}")
            self.is_loaded = True  # Loading was attempted
            self.frame = None
            return None

    def reset(self) -> None:
        """
        Reset the frame state.
        """
        self.frame = None
        self.is_loaded = False
        self.load_attempted = False

Attributes

file_path instance-attribute

metadata = field(default_factory=dict) class-attribute instance-attribute

frame = None class-attribute instance-attribute

is_loaded = False class-attribute instance-attribute

load_attempted = False class-attribute instance-attribute

Functions

__init__(file_path, metadata=dict(), frame=None, is_loaded=False, load_attempted=False)

ensure_loaded(loader)

Ensures the frame is loaded, loading it if necessary.

Parameters:

Name	Type	Description	Default
loader	Callable[[Path], F | None]	Function to load a frame from a file path	required

Returns:

Type	Description
F | None	The loaded frame, or None if loading failed

Source code in wandas/utils/frame_dataset.py

def ensure_loaded(self, loader: Callable[[Path], F | None]) -> F | None:
    """
    Ensures the frame is loaded, loading it if necessary.

    Args:
        loader: Function to load a frame from a file path

    Returns:
        The loaded frame, or None if loading failed
    """
    # Return the current frame if already loaded
    if self.is_loaded:
        return self.frame

    # Attempt to load if not loaded yet
    try:
        self.load_attempted = True
        self.frame = loader(self.file_path)
        self.is_loaded = True
        return self.frame
    except Exception as e:
        logger.error(f"Failed to load file {self.file_path}: {e!s}")
        self.is_loaded = True  # Loading was attempted
        self.frame = None
        return None

reset()

Reset the frame state.

Source code in wandas/utils/frame_dataset.py

def reset(self) -> None:
    """
    Reset the frame state.
    """
    self.frame = None
    self.is_loaded = False
    self.load_attempted = False

FrameDataset

Bases: Generic[F], ABC

Abstract folder-backed collection of lazily loaded Frames.

File discovery does not create Frames. Integer access creates and caches the requested Frame, while its Dask-backed sample data remains lazy until a Frame materialization API such as frame.data is used. A load or transform failure is cached as an attempted item and represented by None; exceptions are also logged.

Dataset transforms create a new dataset and leave the source dataset unchanged. apply(), resample(), trim(), and normalize() preserve the dataset subtype; stft() intentionally returns SpectrogramFrameDataset. Discovered file metadata is deep-copied into derived datasets and attached to each successfully loaded or transformed Frame.

get_metadata() returns current summary state. It does not expose a processing-history or lineage API for the dataset.

Source code in wandas/utils/frame_dataset.py

class FrameDataset(Generic[F], ABC):
    """
    Abstract folder-backed collection of lazily loaded Frames.

    File discovery does not create Frames. Integer access creates and caches the
    requested Frame, while its Dask-backed sample data remains lazy until a Frame
    materialization API such as ``frame.data`` is used. A load or transform failure
    is cached as an attempted item and represented by ``None``; exceptions are also
    logged.

    Dataset transforms create a new dataset and leave the source dataset unchanged.
    ``apply()``, ``resample()``, ``trim()``, and ``normalize()`` preserve the dataset
    subtype; ``stft()`` intentionally returns ``SpectrogramFrameDataset``. Discovered
    file metadata is deep-copied into derived datasets and attached to each
    successfully loaded or transformed Frame.

    ``get_metadata()`` returns current summary state. It does not expose a
    processing-history or lineage API for the dataset.
    """

    def __init__(
        self,
        folder_path: str,
        sampling_rate: int | None = None,
        signal_length: int | None = None,
        file_extensions: list[str] | None = None,
        lazy_loading: bool = True,
        recursive: bool = False,
        source_dataset: "FrameDataset[Any] | None" = None,
        transform: Callable[[Any], F | None] | None = None,
        metadata_resolver: MetadataResolver | None = None,
        path_metadata: bool = False,
    ):
        if path_metadata and metadata_resolver is not None:
            raise ValueError(
                "path_metadata=True cannot be combined with metadata_resolver; "
                "disable one metadata source to avoid ambiguous precedence"
            )
        self.folder_path = Path(folder_path)
        if source_dataset is None and not self.folder_path.exists():
            raise FileNotFoundError(f"Folder does not exist: {self.folder_path}")

        self.sampling_rate = sampling_rate
        self.signal_length = signal_length
        self.file_extensions = file_extensions or [".wav"]
        self._recursive = recursive
        self._lazy_loading = lazy_loading
        self._metadata_resolver = metadata_resolver
        self._path_metadata = path_metadata

        # Changed to a list of LazyFrame
        self._lazy_frames: list[LazyFrame[F]] = []

        self._source_dataset = source_dataset
        self._transform = transform

        if self._source_dataset is not None:
            self._initialize_from_source()
        else:
            self._initialize_from_folder()

    def _initialize_from_source(self) -> None:
        """Initialize from a source dataset."""
        if self._source_dataset is None:
            return

        self._lazy_frames = [
            LazyFrame(lazy_frame.file_path, metadata=deepcopy(lazy_frame.metadata))
            for lazy_frame in self._source_dataset._lazy_frames
        ]

        # Inherit other properties
        self.sampling_rate = self.sampling_rate or self._source_dataset.sampling_rate
        self.signal_length = self.signal_length or self._source_dataset.signal_length
        self.file_extensions = self.file_extensions or self._source_dataset.file_extensions
        self._recursive = self._source_dataset._recursive
        self._path_metadata = self._source_dataset._path_metadata
        self.folder_path = self._source_dataset.folder_path

    def _initialize_from_folder(self) -> None:
        """Initialize from a folder."""
        self._discover_files()
        if not self._lazy_loading:
            self._load_all_files()

    def _discover_files(self) -> None:
        """Discover files in the folder and store them in a list of LazyFrame."""
        file_paths = []
        for ext in self.file_extensions:
            pattern = f"**/*{ext}" if self._recursive else f"*{ext}"
            file_paths.extend(sorted(p for p in self.folder_path.glob(pattern) if p.is_file()))

        # Remove duplicates and sort
        file_paths = sorted(set(file_paths))

        self._lazy_frames = [
            LazyFrame(file_path, metadata=self._resolve_metadata(file_path)) for file_path in file_paths
        ]

    def _resolve_metadata(self, file_path: Path) -> dict[str, object]:
        """Resolve and validate metadata for one discovered file."""
        if not self._path_metadata and self._metadata_resolver is None:
            return {}
        relative_path = file_path.relative_to(self.folder_path)
        if self._path_metadata:
            return self._resolve_path_metadata(relative_path)
        assert self._metadata_resolver is not None
        display_path = relative_path.as_posix()
        try:
            resolved = self._metadata_resolver(relative_path)
        except Exception as exc:
            raise RuntimeError(f"Metadata resolver failed for {display_path}: {exc}") from exc
        if not isinstance(resolved, Mapping):
            raise TypeError(
                f"Metadata resolver must return a mapping for {display_path}; got {type(resolved).__name__}"
            )
        invalid_keys = [key for key in resolved if not isinstance(key, str)]
        if invalid_keys:
            raise TypeError(f"Metadata keys must be strings for {display_path}; got {invalid_keys!r}")
        reserved_keys = _RESERVED_METADATA_KEYS.intersection(resolved)
        if reserved_keys:
            names = ", ".join(sorted(reserved_keys))
            raise ValueError(f"Metadata resolver cannot set reserved key(s) for {display_path}: {names}")
        return deepcopy(dict(resolved))

    @staticmethod
    def _resolve_path_metadata(relative_path: Path) -> dict[str, object]:
        """Infer AWS Glue-style metadata from relative parent segments."""
        metadata: dict[str, object] = {}
        for index, segment in enumerate(relative_path.parent.parts):
            hive_key, separator, hive_value = segment.partition("=")
            if separator and hive_key:
                key, value = hive_key, hive_value
                if key in _RESERVED_METADATA_KEYS:
                    raise ValueError(
                        f"Path metadata cannot set reserved key {key!r} for {relative_path}; "
                        "rename the Hive partition key"
                    )
                if key.startswith("partition_") and key.removeprefix("partition_").isdigit():
                    raise ValueError(
                        f"Hive path metadata key {key!r} uses the generated partition namespace "
                        f"for {relative_path}; rename the Hive partition key"
                    )
            else:
                key, value = f"partition_{index}", segment
            if key in metadata:
                raise ValueError(
                    f"Duplicate path metadata key {key!r} for {relative_path}; "
                    "rename the Hive partition key or directory segment"
                )
            metadata[key] = value
        return metadata

    @staticmethod
    def _attach_metadata(frame: F | None, metadata: Mapping[str, object]) -> F | None:
        """Return a new frame carrying a deep copy of file metadata."""
        if frame is None or not metadata:
            return frame
        merged = deepcopy(dict(frame.metadata))
        merged.update(deepcopy(dict(metadata)))
        return frame._create_new_instance(frame._data, metadata=merged, previous=frame.previous)

    def _load_all_files(self) -> None:
        """Load all files."""
        for i in _progress(range(len(self._lazy_frames)), desc="Loading/transforming"):
            try:
                self._ensure_loaded(i)
            except Exception as e:
                filepath = self._lazy_frames[i].file_path
                logger.warning(f"Failed to load/transform index {i} ({filepath}): {e!s}")
        self._lazy_loading = False

    @abstractmethod
    def _load_file(self, file_path: Path) -> F | None:
        """Abstract method to load a frame from a file."""

    def _load_from_source(self, index: int) -> F | None:
        """Load a frame from the source dataset and transform it if necessary."""
        if self._source_dataset is None or self._transform is None:
            return None

        source_frame = self._source_dataset._ensure_loaded(index)
        if source_frame is None:
            return None

        try:
            return self._transform(source_frame)
        except Exception as e:
            msg = f"Failed to transform index {index}: {e!s}"
            logger.warning(msg)
            # Also emit to the root logger to improve capture reliability
            # in test runners and across different logging configurations
            logging.getLogger().warning(msg)
            return None

    def _ensure_loaded(self, index: int) -> F | None:
        """Load and cache one item, returning ``None`` after a load or transform failure."""
        if not (0 <= index < len(self._lazy_frames)):
            raise IndexError(f"Index {index} is out of range (0-{len(self._lazy_frames) - 1})")

        lazy_frame = self._lazy_frames[index]

        # Return if already loaded
        if lazy_frame.is_loaded:
            return lazy_frame.frame

        try:
            # Convert from source dataset
            if self._transform is not None and self._source_dataset is not None:
                lazy_frame.load_attempted = True
                frame = self._load_from_source(index)
                frame = self._attach_metadata(frame, lazy_frame.metadata)
                lazy_frame.frame = frame
                lazy_frame.is_loaded = True
                return frame
            # Load directly from file
            frame = lazy_frame.ensure_loaded(self._load_file)
            frame = self._attach_metadata(frame, lazy_frame.metadata)
            lazy_frame.frame = frame
            return frame
        except Exception as e:
            f_path = lazy_frame.file_path
            logger.error(f"Failed to load or initialize index {index} ({f_path}): {e!s}")
            lazy_frame.frame = None
            lazy_frame.is_loaded = True
            lazy_frame.load_attempted = True
            return None

    def _get_file_paths(self) -> list[Path]:
        """Get a list of file paths."""
        return [lazy_frame.file_path for lazy_frame in self._lazy_frames]

    def __len__(self) -> int:
        """Return the number of files in the dataset."""
        return len(self._lazy_frames)

    def get_by_label(self, label: str) -> F | None:
        """
        Get a frame by its label (filename).

        Deprecated since 0.2.0. Use ``get_all_by_label()`` instead. The
        first-match behavior is planned for removal no earlier than version
        0.7.0.

        Args:
            label (str): Filename (label) to search for, such as
                ``"sample_1.wav"``.

        Returns:
            F | None: Matching frame, or ``None`` when no frame is found.

        Examples:
            >>> frame = dataset.get_by_label("sample_1.wav")
            >>> if frame:
            ...     print(frame.label)
        """
        warnings.warn(
            "FrameDataset.get_by_label() is deprecated since 0.2.0 and is planned "
            "for removal no earlier than 0.7.0; use get_all_by_label() to obtain "
            "all matches.",
            DeprecationWarning,
            stacklevel=2,
        )
        all_matches = self.get_all_by_label(label)
        if len(all_matches) > 0:
            return all_matches[0]
        return None

    def get_all_by_label(self, label: str) -> list[F]:
        """
        Get all frames matching the given label (filename).

        Args:
            label: str. The filename (label) to search for (e.g., 'sample_1.wav').

        Returns:
            list[F]: A list of frames matching the label.
                If none are found, returns an empty list.

        Notes:
            - Search is performed against the filename portion only (i.e. Path.name).
            - Each matched frame will be loaded (triggering lazy load) via `_ensure_loaded`.
        """
        matches: list[F] = []
        for i, lazy_frame in enumerate(self._lazy_frames):
            if lazy_frame.file_path.name == label:
                loaded = self._ensure_loaded(i)
                if loaded is not None:
                    matches.append(loaded)
        return matches

    @overload
    def __getitem__(self, key: int) -> F | None: ...

    @overload
    def __getitem__(self, key: str) -> list[F]: ...

    def __getitem__(self, key: int | str) -> F | None | list[F]:
        """
        Get the frame by index (int) or label (str).

        Args:
            key: int or str. Index (int) or filename/label (str).

        Returns:
            F | None or list[F]: If ``key`` is an int, returns the cached Frame or ``None`` when loading
                or transformation failed. If ``key`` is a str, returns all successfully
                loaded matching Frames; the list is empty when no match loads.

        Raises:
            IndexError: If an integer index is outside ``0 <= key < len(dataset)``. Negative
                indexing is not supported.
            TypeError: If ``key`` is neither an integer nor a string.

        Notes:
            A ``None`` result is cached as an attempted item and is not retried
            automatically. Exceptions are also logged. Frame creation may inspect a file
            header, but sample values remain Dask-lazy until a Frame materialization API
            is used.

        Examples:
            >>> frame = dataset[0]  # by index
            >>> frames = dataset["sample_1.wav"]  # list of matches by filename
        """
        if isinstance(key, int):
            return self._ensure_loaded(key)
        if isinstance(key, str):
            # pandas-like behaviour: return all matches for the label as a list
            return self.get_all_by_label(key)
        raise TypeError(f"Invalid key type: {type(key)}. Must be int or str.")

    @overload
    def apply(self, func: Callable[[F], F_out | None]) -> "FrameDataset[F_out]": ...

    @overload
    def apply(self, func: Callable[[F], Any | None]) -> "FrameDataset[Any]": ...

    def apply(self, func: Callable[[F], Any | None]) -> "FrameDataset[Any]":
        """Create a lazy transformed dataset without changing this dataset.

        The returned dataset has the same runtime dataset subtype. The callable runs
        once per item when that item is first accessed. Returning ``None`` or raising
        an exception represents a failed/filtered item as ``None``; exceptions are
        logged and do not abort access to other items.

        Discovered file metadata is deep-copied to the derived dataset and attached
        to every successfully transformed Frame. The metadata attached at discovery
        takes precedence over same-named keys returned by ``func``.
        """
        new_dataset = type(self)(
            folder_path=str(self.folder_path),
            lazy_loading=True,
            source_dataset=self,
            transform=func,
            sampling_rate=self.sampling_rate,
            signal_length=self.signal_length,
            file_extensions=self.file_extensions,
            recursive=self._recursive,
        )
        return cast("FrameDataset[Any]", new_dataset)

    def save(self, output_folder: str, filename_prefix: str = "") -> None:
        """Unsupported: dataset-level persistence is not implemented.

        Saving individual Frames is supported through the Frame API. This method is
        retained only to fail explicitly and must not be used as a persistence path.

        Raises:
            NotImplementedError: Always raised because ``FrameDataset.save()`` is unsupported.
        """
        raise NotImplementedError("The save method is not currently implemented.")

    def sample(
        self,
        n: int | None = None,
        ratio: float | None = None,
        seed: int | None = None,
    ) -> "FrameDataset[F]":
        """Return a lazy random subset without loading Frames.

        When both ``n`` and ``ratio`` are omitted, the requested size is
        ``max(1, min(10, int(len(self) * 0.1)))`` and is then capped at
        ``len(self)``. An empty dataset therefore returns an empty subset. When
        ``ratio`` is provided, the requested size is
        ``max(1, int(len(self) * ratio))``; an explicit ``n`` takes precedence over
        ``ratio``. Explicit sizes are also clamped to the inclusive range from one
        to the dataset length for non-empty datasets.

        Sampling preserves file metadata and lazy Frame loading. ``seed`` makes the
        selected file indices reproducible.
        """
        if seed is not None:
            random.seed(seed)

        total = len(self._lazy_frames)
        if total == 0:
            return _SampledFrameDataset(self, [])

        # Determine sample size
        if n is None:
            if ratio is None:
                n = max(1, min(10, int(total * 0.1)))
            else:
                n = max(1, int(total * ratio))
        else:
            n = max(1, n)

        n = min(n, total)

        # Randomly select indices
        sampled_indices = sorted(random.sample(range(total), n))

        return _SampledFrameDataset(self, sampled_indices)

    def select(self, **criteria: object) -> "FrameDataset[F]":
        """Select files by exact-match resolver metadata without loading frames."""
        known_keys = {key for lazy_frame in self._lazy_frames for key in lazy_frame.metadata}
        unknown_keys = set(criteria).difference(known_keys)
        if unknown_keys:
            names = ", ".join(sorted(unknown_keys))
            raise KeyError(f"Unknown file metadata key(s): {names}")
        selected_indices = [
            index
            for index, lazy_frame in enumerate(self._lazy_frames)
            if all(key in lazy_frame.metadata and lazy_frame.metadata[key] == value for key, value in criteria.items())
        ]
        if isinstance(self, _SubsetFrameDataset):
            return _SubsetFrameDataset(self, selected_indices)
        subset = _SubsetFrameDataset(self, selected_indices)
        return cast(
            "FrameDataset[F]",
            type(self)(
                folder_path=str(self.folder_path),
                lazy_loading=True,
                source_dataset=subset,
                transform=lambda frame: frame,
                sampling_rate=self.sampling_rate,
                signal_length=self.signal_length,
                file_extensions=self.file_extensions,
                recursive=self._recursive,
            ),
        )

    def get_metadata(self) -> dict[str, Any]:
        """Return current dataset configuration and load-summary state.

        This call does not load any Frames. ``loaded_count`` counts items whose Frame
        load or transform has been attempted, including failed items cached as
        ``None``. ``has_transform`` reports whether this dataset has one lazy
        transform from a source dataset. The result is a summary, not a dataset
        processing history or Frame lineage.
        """
        actual_sr: int | float | None = self.sampling_rate
        frame_type_name = "Unknown"

        # Count loaded frames
        loaded_count = sum(1 for lazy_frame in self._lazy_frames if lazy_frame.is_loaded)

        # Get metadata from the first frame (if possible)
        first_frame: F | None = None
        if len(self._lazy_frames) > 0:
            try:
                if self._lazy_frames[0].is_loaded:
                    first_frame = self._lazy_frames[0].frame

                if first_frame:
                    actual_sr = getattr(first_frame, "sampling_rate", self.sampling_rate)
                    frame_type_name = type(first_frame).__name__
            except Exception as e:
                logger.warning(f"Error accessing the first frame during metadata retrieval: {e}")

        return {
            "folder_path": str(self.folder_path),
            "file_count": len(self._lazy_frames),
            "loaded_count": loaded_count,
            "target_sampling_rate": self.sampling_rate,
            "actual_sampling_rate": actual_sr,
            "signal_length": self.signal_length,
            "file_extensions": self.file_extensions,
            "lazy_loading": self._lazy_loading,
            "recursive": self._recursive,
            "path_metadata": self._path_metadata,
            "frame_type": frame_type_name,
            "has_transform": self._transform is not None,
            "is_sampled": isinstance(self, _SampledFrameDataset),
        }

Attributes

folder_path = Path(folder_path) instance-attribute

sampling_rate = sampling_rate instance-attribute

signal_length = signal_length instance-attribute

file_extensions = file_extensions or ['.wav'] instance-attribute

Functions

__init__(folder_path, sampling_rate=None, signal_length=None, file_extensions=None, lazy_loading=True, recursive=False, source_dataset=None, transform=None, metadata_resolver=None, path_metadata=False)

Source code in wandas/utils/frame_dataset.py

def __init__(
    self,
    folder_path: str,
    sampling_rate: int | None = None,
    signal_length: int | None = None,
    file_extensions: list[str] | None = None,
    lazy_loading: bool = True,
    recursive: bool = False,
    source_dataset: "FrameDataset[Any] | None" = None,
    transform: Callable[[Any], F | None] | None = None,
    metadata_resolver: MetadataResolver | None = None,
    path_metadata: bool = False,
):
    if path_metadata and metadata_resolver is not None:
        raise ValueError(
            "path_metadata=True cannot be combined with metadata_resolver; "
            "disable one metadata source to avoid ambiguous precedence"
        )
    self.folder_path = Path(folder_path)
    if source_dataset is None and not self.folder_path.exists():
        raise FileNotFoundError(f"Folder does not exist: {self.folder_path}")

    self.sampling_rate = sampling_rate
    self.signal_length = signal_length
    self.file_extensions = file_extensions or [".wav"]
    self._recursive = recursive
    self._lazy_loading = lazy_loading
    self._metadata_resolver = metadata_resolver
    self._path_metadata = path_metadata

    # Changed to a list of LazyFrame
    self._lazy_frames: list[LazyFrame[F]] = []

    self._source_dataset = source_dataset
    self._transform = transform

    if self._source_dataset is not None:
        self._initialize_from_source()
    else:
        self._initialize_from_folder()

__len__()

Return the number of files in the dataset.

Source code in wandas/utils/frame_dataset.py

def __len__(self) -> int:
    """Return the number of files in the dataset."""
    return len(self._lazy_frames)

get_by_label(label)

Get a frame by its label (filename).

Deprecated since 0.2.0. Use get_all_by_label() instead. The first-match behavior is planned for removal no earlier than version 0.7.0.

Parameters:

Name	Type	Description	Default
label	str	Filename (label) to search for, such as "sample_1.wav".	required

Returns:

Type	Description
F | None	F | None: Matching frame, or None when no frame is found.

Examples:

>>> frame = dataset.get_by_label("sample_1.wav")
>>> if frame:
...     print(frame.label)

Source code in wandas/utils/frame_dataset.py

def get_by_label(self, label: str) -> F | None:
    """
    Get a frame by its label (filename).

    Deprecated since 0.2.0. Use ``get_all_by_label()`` instead. The
    first-match behavior is planned for removal no earlier than version
    0.7.0.

    Args:
        label (str): Filename (label) to search for, such as
            ``"sample_1.wav"``.

    Returns:
        F | None: Matching frame, or ``None`` when no frame is found.

    Examples:
        >>> frame = dataset.get_by_label("sample_1.wav")
        >>> if frame:
        ...     print(frame.label)
    """
    warnings.warn(
        "FrameDataset.get_by_label() is deprecated since 0.2.0 and is planned "
        "for removal no earlier than 0.7.0; use get_all_by_label() to obtain "
        "all matches.",
        DeprecationWarning,
        stacklevel=2,
    )
    all_matches = self.get_all_by_label(label)
    if len(all_matches) > 0:
        return all_matches[0]
    return None

get_all_by_label(label)

Get all frames matching the given label (filename).

Parameters:

Name	Type	Description	Default
label	str	str. The filename (label) to search for (e.g., 'sample_1.wav').	required

Returns:

Type	Description
list[F]	list[F]: A list of frames matching the label. If none are found, returns an empty list.

Notes

• Search is performed against the filename portion only (i.e. Path.name).
• Each matched frame will be loaded (triggering lazy load) via _ensure_loaded.

Source code in wandas/utils/frame_dataset.py

def get_all_by_label(self, label: str) -> list[F]:
    """
    Get all frames matching the given label (filename).

    Args:
        label: str. The filename (label) to search for (e.g., 'sample_1.wav').

    Returns:
        list[F]: A list of frames matching the label.
            If none are found, returns an empty list.

    Notes:
        - Search is performed against the filename portion only (i.e. Path.name).
        - Each matched frame will be loaded (triggering lazy load) via `_ensure_loaded`.
    """
    matches: list[F] = []
    for i, lazy_frame in enumerate(self._lazy_frames):
        if lazy_frame.file_path.name == label:
            loaded = self._ensure_loaded(i)
            if loaded is not None:
                matches.append(loaded)
    return matches

__getitem__(key)

__getitem__(key: int) -> F | None

__getitem__(key: str) -> list[F]

Get the frame by index (int) or label (str).

Parameters:

Name	Type	Description	Default
key	int | str	int or str. Index (int) or filename/label (str).	required

Returns:

Type	Description
F | None | list[F]	F | None or list[F]: If key is an int, returns the cached Frame or None when loading or transformation failed. If key is a str, returns all successfully loaded matching Frames; the list is empty when no match loads.

Raises:

Type	Description
IndexError	If an integer index is outside 0 <= key < len(dataset). Negative indexing is not supported.
TypeError	If key is neither an integer nor a string.

Notes

A None result is cached as an attempted item and is not retried automatically. Exceptions are also logged. Frame creation may inspect a file header, but sample values remain Dask-lazy until a Frame materialization API is used.

Examples:

>>> frame = dataset[0]  # by index
>>> frames = dataset["sample_1.wav"]  # list of matches by filename

Source code in wandas/utils/frame_dataset.py

def __getitem__(self, key: int | str) -> F | None | list[F]:
    """
    Get the frame by index (int) or label (str).

    Args:
        key: int or str. Index (int) or filename/label (str).

    Returns:
        F | None or list[F]: If ``key`` is an int, returns the cached Frame or ``None`` when loading
            or transformation failed. If ``key`` is a str, returns all successfully
            loaded matching Frames; the list is empty when no match loads.

    Raises:
        IndexError: If an integer index is outside ``0 <= key < len(dataset)``. Negative
            indexing is not supported.
        TypeError: If ``key`` is neither an integer nor a string.

    Notes:
        A ``None`` result is cached as an attempted item and is not retried
        automatically. Exceptions are also logged. Frame creation may inspect a file
        header, but sample values remain Dask-lazy until a Frame materialization API
        is used.

    Examples:
        >>> frame = dataset[0]  # by index
        >>> frames = dataset["sample_1.wav"]  # list of matches by filename
    """
    if isinstance(key, int):
        return self._ensure_loaded(key)
    if isinstance(key, str):
        # pandas-like behaviour: return all matches for the label as a list
        return self.get_all_by_label(key)
    raise TypeError(f"Invalid key type: {type(key)}. Must be int or str.")

apply(func)

apply(func: Callable[[F], F_out | None]) -> FrameDataset[F_out]

apply(func: Callable[[F], Any | None]) -> FrameDataset[Any]

Create a lazy transformed dataset without changing this dataset.

The returned dataset has the same runtime dataset subtype. The callable runs once per item when that item is first accessed. Returning None or raising an exception represents a failed/filtered item as None; exceptions are logged and do not abort access to other items.

Discovered file metadata is deep-copied to the derived dataset and attached to every successfully transformed Frame. The metadata attached at discovery takes precedence over same-named keys returned by func.

Source code in wandas/utils/frame_dataset.py

def apply(self, func: Callable[[F], Any | None]) -> "FrameDataset[Any]":
    """Create a lazy transformed dataset without changing this dataset.

    The returned dataset has the same runtime dataset subtype. The callable runs
    once per item when that item is first accessed. Returning ``None`` or raising
    an exception represents a failed/filtered item as ``None``; exceptions are
    logged and do not abort access to other items.

    Discovered file metadata is deep-copied to the derived dataset and attached
    to every successfully transformed Frame. The metadata attached at discovery
    takes precedence over same-named keys returned by ``func``.
    """
    new_dataset = type(self)(
        folder_path=str(self.folder_path),
        lazy_loading=True,
        source_dataset=self,
        transform=func,
        sampling_rate=self.sampling_rate,
        signal_length=self.signal_length,
        file_extensions=self.file_extensions,
        recursive=self._recursive,
    )
    return cast("FrameDataset[Any]", new_dataset)

save(output_folder, filename_prefix='')

Unsupported: dataset-level persistence is not implemented.

Saving individual Frames is supported through the Frame API. This method is retained only to fail explicitly and must not be used as a persistence path.

Raises:

Type	Description
NotImplementedError	Always raised because FrameDataset.save() is unsupported.

Source code in wandas/utils/frame_dataset.py

def save(self, output_folder: str, filename_prefix: str = "") -> None:
    """Unsupported: dataset-level persistence is not implemented.

    Saving individual Frames is supported through the Frame API. This method is
    retained only to fail explicitly and must not be used as a persistence path.

    Raises:
        NotImplementedError: Always raised because ``FrameDataset.save()`` is unsupported.
    """
    raise NotImplementedError("The save method is not currently implemented.")

sample(n=None, ratio=None, seed=None)

Return a lazy random subset without loading Frames.

When both n and ratio are omitted, the requested size is max(1, min(10, int(len(self) * 0.1))) and is then capped at len(self). An empty dataset therefore returns an empty subset. When ratio is provided, the requested size is max(1, int(len(self) * ratio)); an explicit n takes precedence over ratio. Explicit sizes are also clamped to the inclusive range from one to the dataset length for non-empty datasets.

Sampling preserves file metadata and lazy Frame loading. seed makes the selected file indices reproducible.

Source code in wandas/utils/frame_dataset.py

def sample(
    self,
    n: int | None = None,
    ratio: float | None = None,
    seed: int | None = None,
) -> "FrameDataset[F]":
    """Return a lazy random subset without loading Frames.

    When both ``n`` and ``ratio`` are omitted, the requested size is
    ``max(1, min(10, int(len(self) * 0.1)))`` and is then capped at
    ``len(self)``. An empty dataset therefore returns an empty subset. When
    ``ratio`` is provided, the requested size is
    ``max(1, int(len(self) * ratio))``; an explicit ``n`` takes precedence over
    ``ratio``. Explicit sizes are also clamped to the inclusive range from one
    to the dataset length for non-empty datasets.

    Sampling preserves file metadata and lazy Frame loading. ``seed`` makes the
    selected file indices reproducible.
    """
    if seed is not None:
        random.seed(seed)

    total = len(self._lazy_frames)
    if total == 0:
        return _SampledFrameDataset(self, [])

    # Determine sample size
    if n is None:
        if ratio is None:
            n = max(1, min(10, int(total * 0.1)))
        else:
            n = max(1, int(total * ratio))
    else:
        n = max(1, n)

    n = min(n, total)

    # Randomly select indices
    sampled_indices = sorted(random.sample(range(total), n))

    return _SampledFrameDataset(self, sampled_indices)

select(**criteria)

Select files by exact-match resolver metadata without loading frames.

Source code in wandas/utils/frame_dataset.py

def select(self, **criteria: object) -> "FrameDataset[F]":
    """Select files by exact-match resolver metadata without loading frames."""
    known_keys = {key for lazy_frame in self._lazy_frames for key in lazy_frame.metadata}
    unknown_keys = set(criteria).difference(known_keys)
    if unknown_keys:
        names = ", ".join(sorted(unknown_keys))
        raise KeyError(f"Unknown file metadata key(s): {names}")
    selected_indices = [
        index
        for index, lazy_frame in enumerate(self._lazy_frames)
        if all(key in lazy_frame.metadata and lazy_frame.metadata[key] == value for key, value in criteria.items())
    ]
    if isinstance(self, _SubsetFrameDataset):
        return _SubsetFrameDataset(self, selected_indices)
    subset = _SubsetFrameDataset(self, selected_indices)
    return cast(
        "FrameDataset[F]",
        type(self)(
            folder_path=str(self.folder_path),
            lazy_loading=True,
            source_dataset=subset,
            transform=lambda frame: frame,
            sampling_rate=self.sampling_rate,
            signal_length=self.signal_length,
            file_extensions=self.file_extensions,
            recursive=self._recursive,
        ),
    )

get_metadata()

Return current dataset configuration and load-summary state.

This call does not load any Frames. loaded_count counts items whose Frame load or transform has been attempted, including failed items cached as None. has_transform reports whether this dataset has one lazy transform from a source dataset. The result is a summary, not a dataset processing history or Frame lineage.

Source code in wandas/utils/frame_dataset.py

def get_metadata(self) -> dict[str, Any]:
    """Return current dataset configuration and load-summary state.

    This call does not load any Frames. ``loaded_count`` counts items whose Frame
    load or transform has been attempted, including failed items cached as
    ``None``. ``has_transform`` reports whether this dataset has one lazy
    transform from a source dataset. The result is a summary, not a dataset
    processing history or Frame lineage.
    """
    actual_sr: int | float | None = self.sampling_rate
    frame_type_name = "Unknown"

    # Count loaded frames
    loaded_count = sum(1 for lazy_frame in self._lazy_frames if lazy_frame.is_loaded)

    # Get metadata from the first frame (if possible)
    first_frame: F | None = None
    if len(self._lazy_frames) > 0:
        try:
            if self._lazy_frames[0].is_loaded:
                first_frame = self._lazy_frames[0].frame

            if first_frame:
                actual_sr = getattr(first_frame, "sampling_rate", self.sampling_rate)
                frame_type_name = type(first_frame).__name__
        except Exception as e:
            logger.warning(f"Error accessing the first frame during metadata retrieval: {e}")

    return {
        "folder_path": str(self.folder_path),
        "file_count": len(self._lazy_frames),
        "loaded_count": loaded_count,
        "target_sampling_rate": self.sampling_rate,
        "actual_sampling_rate": actual_sr,
        "signal_length": self.signal_length,
        "file_extensions": self.file_extensions,
        "lazy_loading": self._lazy_loading,
        "recursive": self._recursive,
        "path_metadata": self._path_metadata,
        "frame_type": frame_type_name,
        "has_transform": self._transform is not None,
        "is_sampled": isinstance(self, _SampledFrameDataset),
    }

ChannelFrameDataset

Bases: FrameDataset[ChannelFrame]

Dataset class for handling audio files as ChannelFrames in a folder.

Source code in wandas/utils/frame_dataset.py

class ChannelFrameDataset(FrameDataset[ChannelFrame]):
    """
    Dataset class for handling audio files as ChannelFrames in a folder.
    """

    def __init__(
        self,
        folder_path: str,
        sampling_rate: int | None = None,
        signal_length: int | None = None,
        file_extensions: list[str] | None = None,
        lazy_loading: bool = True,
        recursive: bool = False,
        source_dataset: "FrameDataset[Any] | None" = None,
        transform: Callable[[Any], ChannelFrame | None] | None = None,
        metadata_resolver: MetadataResolver | None = None,
        path_metadata: bool = False,
    ):
        _file_extensions = file_extensions if file_extensions is not None else supported_formats()

        super().__init__(
            folder_path=folder_path,
            sampling_rate=sampling_rate,
            signal_length=signal_length,
            file_extensions=_file_extensions,
            lazy_loading=lazy_loading,
            recursive=recursive,
            source_dataset=source_dataset,
            transform=transform,
            metadata_resolver=metadata_resolver,
            path_metadata=path_metadata,
        )

    def _load_file(self, file_path: Path) -> ChannelFrame | None:
        """Load an audio file and return a ChannelFrame."""
        try:
            frame = ChannelFrame.from_file(file_path)
            if self.sampling_rate and frame.sampling_rate != self.sampling_rate:
                logger.info(
                    f"Resampling file {file_path.name} ({frame.sampling_rate} Hz) to "
                    f"dataset rate ({self.sampling_rate} Hz)."
                )
                frame = frame.resampling(target_sr=self.sampling_rate)
            return frame
        except Exception as e:
            logger.error(f"Failed to load or initialize file {file_path}: {e!s}")
            return None

    def select(self, **criteria: object) -> "ChannelFrameDataset":
        """Select files while preserving ChannelFrameDataset processing methods."""
        return cast(ChannelFrameDataset, super().select(**criteria))

    def resample(self, target_sr: int) -> "ChannelFrameDataset":
        """Resample all frames in the dataset."""

        def _resample_func(frame: ChannelFrame) -> ChannelFrame | None:
            if frame is None:
                return None
            try:
                return frame.resampling(target_sr=target_sr)
            except Exception as e:
                logger.warning(f"Resampling error (target_sr={target_sr}): {e}")
                return None

        new_dataset = self.apply(_resample_func)
        return cast(ChannelFrameDataset, new_dataset)

    def trim(self, start: float, end: float) -> "ChannelFrameDataset":
        """Trim all frames in the dataset."""

        def _trim_func(frame: ChannelFrame) -> ChannelFrame | None:
            if frame is None:
                return None
            try:
                return frame.trim(start=start, end=end)
            except Exception as e:
                logger.warning(f"Trimming error (start={start}, end={end}): {e}")
                return None

        new_dataset = self.apply(_trim_func)
        return cast(ChannelFrameDataset, new_dataset)

    def normalize(self, **kwargs: Any) -> "ChannelFrameDataset":
        """Normalize all frames in the dataset."""

        def _normalize_func(frame: ChannelFrame) -> ChannelFrame | None:
            if frame is None:
                return None
            try:
                return frame.normalize(**kwargs)
            except Exception as e:
                logger.warning(f"Normalization error ({kwargs}): {e}")
                return None

        new_dataset = self.apply(_normalize_func)
        return cast(ChannelFrameDataset, new_dataset)

    def stft(
        self,
        n_fft: int = 2048,
        hop_length: int | None = None,
        win_length: int | None = None,
        window: str = "hann",
    ) -> "SpectrogramFrameDataset":
        """Apply STFT to all frames in the dataset."""
        _hop = hop_length or n_fft // 4

        def _stft_func(frame: ChannelFrame) -> SpectrogramFrame | None:
            if frame is None:
                return None
            try:
                return frame.stft(
                    n_fft=n_fft,
                    hop_length=_hop,
                    win_length=win_length,
                    window=window,
                )
            except Exception as e:
                logger.warning(f"STFT error (n_fft={n_fft}, hop={_hop}): {e}")
                return None

        new_dataset = SpectrogramFrameDataset(
            folder_path=str(self.folder_path),
            lazy_loading=True,
            source_dataset=self,
            transform=_stft_func,
            sampling_rate=self.sampling_rate,
        )
        return new_dataset

    @classmethod
    def from_folder(
        cls,
        folder_path: str,
        sampling_rate: int | None = None,
        file_extensions: list[str] | None = None,
        recursive: bool = False,
        lazy_loading: bool = True,
        metadata_resolver: MetadataResolver | None = None,
        path_metadata: bool = False,
    ) -> "ChannelFrameDataset":
        """Create a dataset, optionally inferring metadata from parent paths."""
        extensions = file_extensions if file_extensions is not None else supported_formats()

        return cls(
            folder_path,
            sampling_rate=sampling_rate,
            file_extensions=extensions,
            lazy_loading=lazy_loading,
            recursive=recursive,
            metadata_resolver=metadata_resolver,
            path_metadata=path_metadata,
        )

Functions

__init__(folder_path, sampling_rate=None, signal_length=None, file_extensions=None, lazy_loading=True, recursive=False, source_dataset=None, transform=None, metadata_resolver=None, path_metadata=False)

Source code in wandas/utils/frame_dataset.py

def __init__(
    self,
    folder_path: str,
    sampling_rate: int | None = None,
    signal_length: int | None = None,
    file_extensions: list[str] | None = None,
    lazy_loading: bool = True,
    recursive: bool = False,
    source_dataset: "FrameDataset[Any] | None" = None,
    transform: Callable[[Any], ChannelFrame | None] | None = None,
    metadata_resolver: MetadataResolver | None = None,
    path_metadata: bool = False,
):
    _file_extensions = file_extensions if file_extensions is not None else supported_formats()

    super().__init__(
        folder_path=folder_path,
        sampling_rate=sampling_rate,
        signal_length=signal_length,
        file_extensions=_file_extensions,
        lazy_loading=lazy_loading,
        recursive=recursive,
        source_dataset=source_dataset,
        transform=transform,
        metadata_resolver=metadata_resolver,
        path_metadata=path_metadata,
    )

select(**criteria)

Select files while preserving ChannelFrameDataset processing methods.

Source code in wandas/utils/frame_dataset.py

def select(self, **criteria: object) -> "ChannelFrameDataset":
    """Select files while preserving ChannelFrameDataset processing methods."""
    return cast(ChannelFrameDataset, super().select(**criteria))

resample(target_sr)

Resample all frames in the dataset.

Source code in wandas/utils/frame_dataset.py

def resample(self, target_sr: int) -> "ChannelFrameDataset":
    """Resample all frames in the dataset."""

    def _resample_func(frame: ChannelFrame) -> ChannelFrame | None:
        if frame is None:
            return None
        try:
            return frame.resampling(target_sr=target_sr)
        except Exception as e:
            logger.warning(f"Resampling error (target_sr={target_sr}): {e}")
            return None

    new_dataset = self.apply(_resample_func)
    return cast(ChannelFrameDataset, new_dataset)

trim(start, end)

Trim all frames in the dataset.

Source code in wandas/utils/frame_dataset.py

def trim(self, start: float, end: float) -> "ChannelFrameDataset":
    """Trim all frames in the dataset."""

    def _trim_func(frame: ChannelFrame) -> ChannelFrame | None:
        if frame is None:
            return None
        try:
            return frame.trim(start=start, end=end)
        except Exception as e:
            logger.warning(f"Trimming error (start={start}, end={end}): {e}")
            return None

    new_dataset = self.apply(_trim_func)
    return cast(ChannelFrameDataset, new_dataset)

normalize(**kwargs)

Normalize all frames in the dataset.

Source code in wandas/utils/frame_dataset.py

def normalize(self, **kwargs: Any) -> "ChannelFrameDataset":
    """Normalize all frames in the dataset."""

    def _normalize_func(frame: ChannelFrame) -> ChannelFrame | None:
        if frame is None:
            return None
        try:
            return frame.normalize(**kwargs)
        except Exception as e:
            logger.warning(f"Normalization error ({kwargs}): {e}")
            return None

    new_dataset = self.apply(_normalize_func)
    return cast(ChannelFrameDataset, new_dataset)

stft(n_fft=2048, hop_length=None, win_length=None, window='hann')

Apply STFT to all frames in the dataset.

Source code in wandas/utils/frame_dataset.py

def stft(
    self,
    n_fft: int = 2048,
    hop_length: int | None = None,
    win_length: int | None = None,
    window: str = "hann",
) -> "SpectrogramFrameDataset":
    """Apply STFT to all frames in the dataset."""
    _hop = hop_length or n_fft // 4

    def _stft_func(frame: ChannelFrame) -> SpectrogramFrame | None:
        if frame is None:
            return None
        try:
            return frame.stft(
                n_fft=n_fft,
                hop_length=_hop,
                win_length=win_length,
                window=window,
            )
        except Exception as e:
            logger.warning(f"STFT error (n_fft={n_fft}, hop={_hop}): {e}")
            return None

    new_dataset = SpectrogramFrameDataset(
        folder_path=str(self.folder_path),
        lazy_loading=True,
        source_dataset=self,
        transform=_stft_func,
        sampling_rate=self.sampling_rate,
    )
    return new_dataset

from_folder(folder_path, sampling_rate=None, file_extensions=None, recursive=False, lazy_loading=True, metadata_resolver=None, path_metadata=False) classmethod

Create a dataset, optionally inferring metadata from parent paths.

Source code in wandas/utils/frame_dataset.py

@classmethod
def from_folder(
    cls,
    folder_path: str,
    sampling_rate: int | None = None,
    file_extensions: list[str] | None = None,
    recursive: bool = False,
    lazy_loading: bool = True,
    metadata_resolver: MetadataResolver | None = None,
    path_metadata: bool = False,
) -> "ChannelFrameDataset":
    """Create a dataset, optionally inferring metadata from parent paths."""
    extensions = file_extensions if file_extensions is not None else supported_formats()

    return cls(
        folder_path,
        sampling_rate=sampling_rate,
        file_extensions=extensions,
        lazy_loading=lazy_loading,
        recursive=recursive,
        metadata_resolver=metadata_resolver,
        path_metadata=path_metadata,
    )

SpectrogramFrameDataset

Bases: FrameDataset[SpectrogramFrame]

Dataset class for handling spectrogram data as SpectrogramFrames. Expected to be generated mainly as a result of ChannelFrameDataset.stft().

Source code in wandas/utils/frame_dataset.py

class SpectrogramFrameDataset(FrameDataset[SpectrogramFrame]):
    """
    Dataset class for handling spectrogram data as SpectrogramFrames.
    Expected to be generated mainly as a result of ChannelFrameDataset.stft().
    """

    def __init__(
        self,
        folder_path: str,
        sampling_rate: int | None = None,
        signal_length: int | None = None,
        file_extensions: list[str] | None = None,
        lazy_loading: bool = True,
        recursive: bool = False,
        source_dataset: "FrameDataset[Any] | None" = None,
        transform: Callable[[Any], SpectrogramFrame | None] | None = None,
    ):
        super().__init__(
            folder_path=folder_path,
            sampling_rate=sampling_rate,
            signal_length=signal_length,
            file_extensions=file_extensions,
            lazy_loading=lazy_loading,
            recursive=recursive,
            source_dataset=source_dataset,
            transform=transform,
        )

    def _load_file(self, file_path: Path) -> SpectrogramFrame | None:
        """Direct loading from files is not currently supported."""
        logger.warning(
            "No method defined for directly loading SpectrogramFrames. Normally "
            "created from ChannelFrameDataset.stft()."
        )
        raise NotImplementedError("No method defined for directly loading SpectrogramFrames")

    def plot(self, index: int, **kwargs: Any) -> None:
        """Plot the spectrogram at the specified index."""
        try:
            frame = self._ensure_loaded(index)

            if frame is None:
                logger.warning(f"Cannot plot index {index} as it failed to load/transform.")
                return

            plot_method = getattr(frame, "plot", None)
            if callable(plot_method):
                plot_method(**kwargs)
            else:
                logger.warning(
                    f"Frame (index {index}, type {type(frame).__name__}) does not have a plot method implemented."
                )
        except Exception as e:
            logger.error(f"An error occurred while plotting index {index}: {e}")

Functions

__init__(folder_path, sampling_rate=None, signal_length=None, file_extensions=None, lazy_loading=True, recursive=False, source_dataset=None, transform=None)

Source code in wandas/utils/frame_dataset.py

def __init__(
    self,
    folder_path: str,
    sampling_rate: int | None = None,
    signal_length: int | None = None,
    file_extensions: list[str] | None = None,
    lazy_loading: bool = True,
    recursive: bool = False,
    source_dataset: "FrameDataset[Any] | None" = None,
    transform: Callable[[Any], SpectrogramFrame | None] | None = None,
):
    super().__init__(
        folder_path=folder_path,
        sampling_rate=sampling_rate,
        signal_length=signal_length,
        file_extensions=file_extensions,
        lazy_loading=lazy_loading,
        recursive=recursive,
        source_dataset=source_dataset,
        transform=transform,
    )

plot(index, **kwargs)

Plot the spectrogram at the specified index.

Source code in wandas/utils/frame_dataset.py

def plot(self, index: int, **kwargs: Any) -> None:
    """Plot the spectrogram at the specified index."""
    try:
        frame = self._ensure_loaded(index)

        if frame is None:
            logger.warning(f"Cannot plot index {index} as it failed to load/transform.")
            return

        plot_method = getattr(frame, "plot", None)
        if callable(plot_method):
            plot_method(**kwargs)
        else:
            logger.warning(
                f"Frame (index {index}, type {type(frame).__name__}) does not have a plot method implemented."
            )
    except Exception as e:
        logger.error(f"An error occurred while plotting index {index}: {e}")

Functions

wandas.utils.generate_sample

Attributes

Frequency = int | float | np.integer[Any] | np.floating[Any] module-attribute

Frequencies = Frequency | list[Any] module-attribute

Classes

Functions

generate_sin(freqs=1000.0, sampling_rate=16000, duration=1.0, label=None)

Generate sample sine wave signals.

Parameters:

Name	Type	Description	Default
freqs	Frequencies	real number or list of real numbers, default=1000.0. Positive finite frequency of each sine wave in Hz. A scalar creates one channel; a list creates one channel per element. Python and NumPy integer and floating scalars are accepted and normalized to float.	1000.0
sampling_rate	int	int, default=16000. Sampling rate in Hz.	16000
duration	float	float, default=1.0. Duration of the signal in seconds.	1.0
label	str | None	str, optional. Label for the entire signal.	None

Returns:

Name	Type	Description
ChannelFrame	ChannelFrame	Dask-backed ChannelFrame containing the sine wave(s).

Raises:

Type	Description
TypeError	If freqs or one of its elements is not a real numeric scalar.
ValueError	If a frequency list is empty or a frequency is non-finite or not positive.

Examples:

>>> import wandas as wd
>>> signal = wd.generate_sin()
>>> signal.sampling_rate
16000

Source code in wandas/utils/generate_sample.py

def generate_sin(
    freqs: Frequencies = 1000.0,
    sampling_rate: int = 16000,
    duration: float = 1.0,
    label: str | None = None,
) -> "ChannelFrame":
    """
    Generate sample sine wave signals.

    Args:
        freqs: real number or list of real numbers, default=1000.0. Positive
            finite frequency of each sine wave in Hz. A scalar creates one
            channel; a list creates one channel per element. Python and NumPy integer
            and floating scalars are accepted and normalized to ``float``.
        sampling_rate: int, default=16000. Sampling rate in Hz.
        duration: float, default=1.0. Duration of the signal in seconds.
        label: str, optional. Label for the entire signal.

    Returns:
        ChannelFrame: Dask-backed ChannelFrame containing the sine wave(s).

    Raises:
        TypeError: If ``freqs`` or one of its elements is not a real numeric scalar.
        ValueError: If a frequency list is empty or a frequency is non-finite or not positive.

    Examples:
        >>> import wandas as wd
        >>> signal = wd.generate_sin()
        >>> signal.sampling_rate
        16000
    """
    return generate_sin_lazy(freqs=freqs, sampling_rate=sampling_rate, duration=duration, label=label)

generate_sin_lazy(freqs=1000.0, sampling_rate=16000, duration=1.0, label=None)

Generate sample sine wave signals using lazy computation.

Parameters:

Name	Type	Description	Default
freqs	Frequencies	real number or list of real numbers, default=1000.0. Positive finite frequency of each sine wave in Hz. A scalar creates one channel; a list creates one channel per element. Python and NumPy integer and floating scalars are accepted and normalized to float.	1000.0
sampling_rate	int	int, default=16000. Sampling rate in Hz.	16000
duration	float	float, default=1.0. Duration of the signal in seconds.	1.0
label	str | None	str, optional. Label for the entire signal.	None

Returns:

Name	Type	Description
ChannelFrame	ChannelFrame	Dask-backed ChannelFrame containing the sine wave(s).

Raises:

Type	Description
TypeError	If freqs or one of its elements is not a real numeric scalar.
ValueError	If a frequency list is empty or a frequency is non-finite or not positive.

Notes

This is the low-level implementation name used by generate_sin. It is not exported from the top-level wandas namespace.

Source code in wandas/utils/generate_sample.py

def generate_sin_lazy(
    freqs: Frequencies = 1000.0,
    sampling_rate: int = 16000,
    duration: float = 1.0,
    label: str | None = None,
) -> "ChannelFrame":
    """
    Generate sample sine wave signals using lazy computation.

    Args:
        freqs: real number or list of real numbers, default=1000.0. Positive
            finite frequency of each sine wave in Hz. A scalar creates one
            channel; a list creates one channel per element. Python and NumPy integer
            and floating scalars are accepted and normalized to ``float``.
        sampling_rate: int, default=16000. Sampling rate in Hz.
        duration: float, default=1.0. Duration of the signal in seconds.
        label: str, optional. Label for the entire signal.

    Returns:
        ChannelFrame: Dask-backed ChannelFrame containing the sine wave(s).

    Raises:
        TypeError: If ``freqs`` or one of its elements is not a real numeric scalar.
        ValueError: If a frequency list is empty or a frequency is non-finite or not positive.

    Notes:
        This is the low-level implementation name used by ``generate_sin``. It is not
        exported from the top-level ``wandas`` namespace.
    """
    from wandas.frames.channel import ChannelFrame

    label = label or "Generated Sin"
    normalized_freqs = _normalize_frequencies(freqs)
    t = np.linspace(0, duration, int(sampling_rate * duration), endpoint=False)

    channels = []
    labels = []
    for idx, freq in enumerate(normalized_freqs):
        data = np.sin(2 * np.pi * freq * t)
        labels.append(f"Channel {idx + 1}")
        channels.append(data)
    return ChannelFrame.from_numpy(
        data=np.array(channels),
        label=label,
        sampling_rate=sampling_rate,
        ch_labels=labels,
    )

wandas.utils.types

Attributes

Real = np.number[Any] module-attribute

Complex = np.complexfloating[Any, Any] module-attribute

NDArrayReal = npt.NDArray[Real] module-attribute

NDArrayComplex = npt.NDArray[Complex] module-attribute

wandas.utils.util

Attributes

DB_FLOOR = 1e-12 module-attribute

PA_REFERENCE = 2e-05 module-attribute

DB_AMIN = 1e-15 module-attribute

Functions

ref_weighted_dB(data, channel_metadata, ndim)

Compute dB level relative to per-channel reference values.

Parameters:

Name	Type	Description	Default
data	NDArrayReal	NDArrayReal. Non-negative amplitude data (already absolute-valued if complex).	required
channel_metadata	list[Any]	list. Objects with a .ref attribute (one per channel).	required
ndim	int	int. Number of dimensions in the underlying dask array.	required

Returns:

Name	Type	Description
NDArrayReal	NDArrayReal	Decibel values: 20 * log10(max(data / ref, DB_FLOOR))

Source code in wandas/utils/util.py

def ref_weighted_dB(
    data: NDArrayReal,
    channel_metadata: list[Any],
    ndim: int,
) -> NDArrayReal:
    """Compute dB level relative to per-channel reference values.

    Args:
        data: NDArrayReal. Non-negative amplitude data (already absolute-valued if complex).
        channel_metadata: list. Objects with a ``.ref`` attribute (one per channel).
        ndim: int. Number of dimensions in the underlying dask array.

    Returns:
        NDArrayReal: Decibel values: ``20 * log10(max(data / ref, DB_FLOOR))``
    """
    ref = np.array([ch.ref for ch in channel_metadata])
    extra_dims = ndim - 1
    ref_shape = ref.reshape((-1,) + (1,) * extra_dims)
    from wandas.processing.weighting import _reference_level_db

    return _reference_level_db(data, ref_shape)

validate_sampling_rate(sampling_rate, param_name='sampling_rate')

Validate that sampling rate is positive.

Parameters:

Name	Type	Description	Default
sampling_rate	float	float. Sampling rate in Hz to validate.	required
param_name	str	str, default="sampling_rate". Name of the parameter being validated (for error messages).	'sampling_rate'

Raises:

Type	Description
ValueError	If sampling_rate is not positive (i.e., <= 0).

Examples:

>>> validate_sampling_rate(44100)  # No error
>>> validate_sampling_rate(0)  # Raises ValueError
>>> validate_sampling_rate(-100)  # Raises ValueError

Source code in wandas/utils/util.py

def validate_sampling_rate(sampling_rate: float, param_name: str = "sampling_rate") -> None:
    """
    Validate that sampling rate is positive.

    Args:
        sampling_rate: float. Sampling rate in Hz to validate.
        param_name: str, default="sampling_rate". Name of the parameter being validated (for error messages).

    Raises:
        ValueError: If sampling_rate is not positive (i.e., <= 0).

    Examples:
        >>> validate_sampling_rate(44100)  # No error
        >>> validate_sampling_rate(0)  # Raises ValueError
        >>> validate_sampling_rate(-100)  # Raises ValueError
    """
    _normalize_sampling_rate(sampling_rate, param_name)

unit_to_ref(unit)

Convert unit to reference value.

Parameters:

Name	Type	Description	Default
unit	str	str. Unit string.	required

Returns:

Name	Type	Description
float	float	Reference value for the unit. For 'Pa', returns 2e-5 (20 μPa). For other units, returns 1.0.

Source code in wandas/utils/util.py

def unit_to_ref(unit: str) -> float:
    """
    Convert unit to reference value.

    Args:
        unit: str. Unit string.

    Returns:
        float: Reference value for the unit. For 'Pa', returns 2e-5 (20 μPa).
            For other units, returns 1.0.
    """
    if unit == "Pa":
        return PA_REFERENCE

    return 1.0

calculate_rms(wave)

Calculate the root mean square of the wave.

Parameters:

Name	Type	Description	Default
wave	NDArrayReal	NDArrayReal. Input waveform data. Can be multi-channel (shape: [channels, samples]) or single channel (shape: [samples]).	required

Returns:

Type	Description
NDArrayReal	Union[float, NDArray[np.float64]]: RMS value(s). For multi-channel input, returns an array of RMS values, one per channel. For single-channel input, returns a single RMS value.

Source code in wandas/utils/util.py

def calculate_rms(wave: "NDArrayReal") -> "NDArrayReal":
    """
    Calculate the root mean square of the wave.

    Args:
        wave: NDArrayReal. Input waveform data. Can be multi-channel (shape: [channels, samples])
            or single channel (shape: [samples]).

    Returns:
        Union[float, NDArray[np.float64]]: RMS value(s). For multi-channel input, returns an array of RMS values,
            one per channel. For single-channel input, returns a single RMS value.
    """
    # Calculate RMS considering axis (over the last dimension)
    axis_to_use = -1 if wave.ndim > 1 else None
    rms_values: NDArrayReal = np.sqrt(np.mean(np.square(wave), axis=axis_to_use, keepdims=True))
    return rms_values

calculate_desired_noise_rms(clean_rms, snr)

Calculate the desired noise RMS based on clean signal RMS and target SNR.

Parameters:

Name	Type	Description	Default
clean_rms	NDArrayReal	"NDArrayReal". RMS value(s) of the clean signal. Can be a single value or an array for multi-channel.	required
snr	float	float. Target Signal-to-Noise Ratio in dB.	required

Returns:

Type	Description
NDArrayReal	"NDArrayReal": Desired noise RMS value(s) to achieve the target SNR.

Source code in wandas/utils/util.py

def calculate_desired_noise_rms(clean_rms: "NDArrayReal", snr: float) -> "NDArrayReal":
    """
    Calculate the desired noise RMS based on clean signal RMS and target SNR.

    Args:
        clean_rms: "NDArrayReal". RMS value(s) of the clean signal.
            Can be a single value or an array for multi-channel.
        snr: float. Target Signal-to-Noise Ratio in dB.

    Returns:
        "NDArrayReal": Desired noise RMS value(s) to achieve the target SNR.
    """
    a = snr / 20
    noise_rms = clean_rms / (10**a)
    return noise_rms

amplitude_to_db(amplitude, ref)

Convert amplitude to decibel.

Parameters:

Name	Type	Description	Default
amplitude	NDArrayReal	NDArrayReal. Input amplitude data.	required
ref	float	float. Reference value for conversion.	required

Returns:

Name	Type	Description
NDArrayReal	NDArrayReal	Amplitude data converted to decibels.

Source code in wandas/utils/util.py

def amplitude_to_db(amplitude: "NDArrayReal", ref: float) -> "NDArrayReal":
    """
    Convert amplitude to decibel.

    Args:
        amplitude: NDArrayReal. Input amplitude data.
        ref: float. Reference value for conversion.

    Returns:
        NDArrayReal: Amplitude data converted to decibels.
    """
    magnitude = np.abs(amplitude)
    ref_magnitude = abs(ref)
    db: NDArrayReal = 20.0 * np.log10(np.maximum(DB_AMIN, magnitude)) - 20.0 * np.log10(max(DB_AMIN, ref_magnitude))
    return db

level_trigger(data, level, offset=0, hold=1)

Find points where the signal crosses the specified level from below.

Parameters:

Name	Type	Description	Default
data	NDArrayReal	NDArrayReal. Input signal data.	required
level	float	float. Threshold level for triggering.	required
offset	int	int, default=0. Offset to add to trigger points.	0
hold	int	int, default=1. Minimum number of samples between successive trigger points.	1

Returns:

Type	Description
list[int]	list of int: List of sample indices where the signal crosses the level.

Source code in wandas/utils/util.py

def level_trigger(data: "NDArrayReal", level: float, offset: int = 0, hold: int = 1) -> list[int]:
    """
    Find points where the signal crosses the specified level from below.

    Args:
        data: NDArrayReal. Input signal data.
        level: float. Threshold level for triggering.
        offset: int, default=0. Offset to add to trigger points.
        hold: int, default=1. Minimum number of samples between successive trigger points.

    Returns:
        list of int: List of sample indices where the signal crosses the level.
    """
    trig_point: list[int] = []

    sig_len = len(data)
    diff = np.diff(np.sign(data - level))
    level_point = np.where(diff > 0)[0]
    level_point = level_point[(level_point + hold) < sig_len]

    if len(level_point) == 0:
        return []

    last_point = level_point[0]
    trig_point.append(last_point + offset)
    for i in level_point:
        if (last_point + hold) < i:
            trig_point.append(i + offset)
            last_point = i

    return trig_point

cut_sig(data, point_list, cut_len, taper_rate=0, dc_cut=False)

Cut segments from signal at specified points.

Parameters:

Name	Type	Description	Default
data	NDArrayReal	NDArrayReal. Input signal data.	required
point_list	list[int]	list of int. List of starting points for cutting.	required
cut_len	int	int. Length of each segment to cut.	required
taper_rate	float	float, default=0. Taper rate for Tukey window applied to segments. A value of 0 means no tapering, 1 means full tapering.	0
dc_cut	bool	bool, default=False. Whether to remove DC component (mean) from segments.	False

Returns:

Name	Type	Description
NDArrayReal	NDArrayReal	Array containing cut segments with shape (n_segments, cut_len).

Source code in wandas/utils/util.py

def cut_sig(
    data: "NDArrayReal",
    point_list: list[int],
    cut_len: int,
    taper_rate: float = 0,
    dc_cut: bool = False,
) -> "NDArrayReal":
    """
    Cut segments from signal at specified points.

    Args:
        data: NDArrayReal. Input signal data.
        point_list: list of int. List of starting points for cutting.
        cut_len: int. Length of each segment to cut.
        taper_rate: float, default=0. Taper rate for Tukey window applied to segments.
            A value of 0 means no tapering, 1 means full tapering.
        dc_cut: bool, default=False. Whether to remove DC component (mean) from segments.

    Returns:
        NDArrayReal: Array containing cut segments with shape (n_segments, cut_len).
    """
    length = len(data)
    point_list_ = [p for p in point_list if p >= 0 and p + cut_len <= length]
    trial: NDArrayReal = np.zeros((len(point_list_), cut_len))

    for i, v in enumerate(point_list_):
        trial[i] = data[v : v + cut_len]
        if dc_cut:
            trial[i] = trial[i] - trial[i].mean()

    win: NDArrayReal = tukey(cut_len, taper_rate).astype(trial.dtype)[np.newaxis, :]
    trial = trial * win
    return trial