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

The wandas.utils module provides various utility functions used in the Wandas library. wandas.utils モジュールは、Wandasライブラリで使用される様々なユーティリティ機能を提供します。

Frame Dataset / フレームデータセット

Provides dataset utilities for managing multiple data frames. 複数のデータフレームを管理するためのデータセットユーティリティを提供します。

Overview / 概要

The FrameDataset classes enable efficient batch processing of audio files in a folder. Key features include: FrameDataset クラスは、フォルダ内の音声ファイルの効率的なバッチ処理を可能にします。主な機能：

• Lazy Loading: Load files only when accessed, reducing memory usage. 遅延読み込み: アクセス時のみファイルを読み込み、メモリ使用量を削減。
• Transformation Chaining: Apply multiple processing operations efficiently. 変換のチェーン: 複数の処理操作を効率的に適用。
• Sampling: Extract random subsets for testing or analysis. サンプリング: テストや分析のためにランダムなサブセットを抽出。
• Metadata Tracking: Keep track of dataset properties and processing history. メタデータ追跡: データセットのプロパティと処理履歴を記録。

Main Classes / 主なクラス

• ChannelFrameDataset: For time-domain data (WAV, FLAC, OGG, AIFF, SND, and CSV files). ChannelFrameDataset: 時間領域データ用（WAV、FLAC、OGG、AIFF、SND、CSVファイル）。
• SpectrogramFrameDataset: For time-frequency domain data (typically created from STFT). SpectrogramFrameDataset: 時間周波数領域データ用（通常はSTFTから作成）。

Basic Usage / 基本的な使用方法

from wandas.utils.frame_dataset import ChannelFrameDataset

# Create a dataset from a folder
# フォルダからデータセットを作成
dataset = ChannelFrameDataset.from_folder(
    folder_path="path/to/audio/files",
    sampling_rate=16000,  # Optional: resample all files to this rate / オプション: すべてのファイルをこのレートにリサンプリング
    file_extensions=[".wav", ".flac"],  # File types to include / 含めるファイルタイプ
    recursive=True,  # Search subdirectories / サブディレクトリを検索
    lazy_loading=True  # Load files on demand (recommended) / オンデマンドでファイルを読み込む（推奨）
)

# Access individual files
# 個別のファイルにアクセス
first_file = dataset[0]
print(f"File: {first_file.label}")
print(f"Duration: {first_file.duration}s")

# Get dataset information
# データセット情報を取得
metadata = dataset.get_metadata()
print(f"Total files: {metadata['file_count']}")
print(f"Loaded files: {metadata['loaded_count']}")

Sampling / サンプリング

Extract random subsets of the dataset for testing or analysis: テストや分析のためにデータセットのランダムなサブセットを抽出：

# Sample by number of files
# ファイル数でサンプリング
sampled = dataset.sample(n=10, seed=42)

# Sample by ratio
# 比率でサンプリング
sampled = dataset.sample(ratio=0.1, seed=42)

# Default: 10% or minimum 1 file
# デフォルト: 10% または最低1ファイル
sampled = dataset.sample(seed=42)

Metadata-driven file selection / メタデータ駆動のファイル選択

Path-derived partitions / パス由来のパーティション

For common partitioned folders, set path_metadata=True instead of writing a resolver. Wandas inspects only each discovered relative path, so discovery remains lazy and does not open audio files. 一般的なパーティションフォルダでは resolver を書かずに path_metadata=True を指定できます。Wandas は探索した相対パスだけを調べるため、探索は遅延性を保ち、音声ファイルを開きません。

dataset = wd.from_folder(
    "recordings/",
    recursive=True,
    path_metadata=True,
)
selected = dataset.select(partition_0="group_a", partition_1="batch_01")

The naming follows AWS Glue crawler conventions: 命名は AWS Glue crawler の規則に従います。

• Plain parent segments use their zero-based path positions: group_a/batch_01/file.wav becomes {"partition_0": "group_a", "partition_1": "batch_01"}. / 通常の親セグメントにはゼロ始まりのパス位置を使います。
• Hive-style segments use their keys: group=group_a/batch=batch_01/file.wav becomes {"group": "group_a", "batch": "batch_01"}. / Hive 形式のセグメントにはそのキーを使います。
• The root folder and filename are excluded. Files at uneven depths receive only the keys present in their own parent path; a missing key does not match a select() criterion. / ルートフォルダとファイル名は除外します。深さが異なる場合は各ファイルに存在するキーだけを持ち、不足キーは select() 条件に一致しません。
• Duplicate Hive keys, the reserved Hive key _source_file, and Hive keys in the generated partition_<number> namespace raise ValueError instead of overwriting or impersonating metadata. / Hive キーの重複、予約済み Hive キー _source_file、生成用 partition_<number> 名前空間の Hive キーは、メタデータの上書きや偽装をせず ValueError になります。

path_metadata=True and metadata_resolver are mutually exclusive because combining two metadata sources would make precedence ambiguous. 2つのメタデータ源の優先順位が曖昧になるため、path_metadata=True と metadata_resolver は同時に指定できません。

Custom metadata resolver / カスタムメタデータ resolver

metadata_resolver receives each path relative to folder_path once during file discovery. This makes it possible to select files without reading audio headers or waveform samples. Multiple select() criteria use exact-match AND semantics. metadata_resolver はファイル探索時に、folder_path からの相対パスを各ファイルにつき一度受け取ります。音声ヘッダーや波形サンプルを読まずにファイルを選択でき、複数の select() 条件は完全一致の AND として扱われます。

Use a custom resolver when metadata comes from a project-specific filename rule rather than parent folders. For ordinary partition folders, prefer path_metadata=True above. 親フォルダではなくプロジェクト固有のファイル名規則からメタデータを得る場合に custom resolver を使います。通常のパーティションフォルダでは、上記の path_metadata=True を優先してください。

from pathlib import Path
import wandas as wd

def resolve_recording_metadata(path: Path):
    filename = path.name
    recording_id, status = filename.removesuffix(".wav").split("__")
    return {
        "recording_id": recording_id,
        "status": status,
    }

dataset = wd.from_folder(
    "recordings/",
    recursive=True,
    file_extensions=[".wav"],
    metadata_resolver=resolve_recording_metadata,
)
selected = dataset.select(status="approved")

Sidecar CSV data remains application-owned in v1. Convert it to a lookup and reference that lookup from the same resolver contract. Specify WAV explicitly because reading a signal CSV currently materializes the whole CSV during header inspection. v1 では sidecar CSV は利用側で管理します。lookup に変換し、同じ resolver 契約から参照してください。信号 CSV のヘッダー確認では現在 CSV 全体を実体化するため、WAV を明示的に指定します。

import pandas as pd

recordings = pd.read_csv("recordings.csv")
lookup = recordings.set_index("path")[["condition", "priority"]].to_dict(orient="index")

dataset = wd.from_folder(
    "recordings/",
    recursive=True,
    file_extensions=[".wav"],
    metadata_resolver=lambda path: lookup[path.as_posix()],
)
selected = dataset.select(condition="reference", priority=1)

Indexing the lookup is recommended because a WAV missing from the CSV fails during dataset construction. Use lookup.get(path.as_posix(), {}) only when files without CSV metadata are intentionally allowed. CSVにないWAVをDataset構築時のエラーにできるため、lookupの添字アクセスを推奨します。CSVメタデータのないファイルも意図的に許可する場合だけ lookup.get(path.as_posix(), {}) を使用します。

Loading stages / 読み込みの3段階

1. from_folder(): discover paths and resolve metadata; no audio headers or waveform samples are read. / パス探索とメタデータ解決。音声ヘッダーと波形は未読。
2. dataset[i]: inspect the audio header and create a Frame; waveform samples remain Dask-lazy. / 音声ヘッダーを確認してFrameを作成。波形はDask遅延のまま。
3. frame.data: materialize waveform samples. / 波形サンプルを実体化。

With lazy_loading=False, stage 2 runs for every file during construction, while stage 3 remains lazy. lazy_loading=False では構築時に全ファイルの段階2を実行しますが、段階3は引き続き遅延します。

Resolver and selection contracts / resolverと選択の契約

• The resolver receives a root-relative Path once per discovered file and must return a Mapping[str, object]. Wandas deep-copies the result. / resolverは探索した各ファイルにつき一度、ルート相対のPathを受け取り、Mapping[str, object]を返します。結果はディープコピーされます。
• Resolver exceptions, non-mapping results, non-string keys, and the reserved _source_file key fail during dataset construction with the affected path. / resolver例外、Mapping以外の戻り値、文字列以外のキー、予約キー_source_fileは対象パスを含む構築時エラーになります。
• Multiple select() criteria use exact-match AND semantics. select() with no criteria returns an independent all-file view. / 複数条件は完全一致のANDです。条件なしのselect()は全ファイルを保持する独立ビューです。
• An unknown key raises KeyError; no matches produce a valid empty Dataset. / 未知キーはKeyError、一致なしは有効な空Datasetです。

Transformations / 変換

Apply processing operations to all files in the dataset: データセット内のすべてのファイルに処理操作を適用：

# Built-in transformations
# 組み込みの変換
resampled = dataset.resample(target_sr=8000)
trimmed = dataset.trim(start=0.5, end=2.0)

# Chain multiple transformations
# 複数の変換をチェーン
processed = (
    dataset
    .resample(target_sr=8000)
    .trim(start=0.5, end=2.0)
)

# Custom transformation
# カスタム変換
def custom_filter(frame):
    return frame.low_pass_filter(cutoff=1000)

filtered = dataset.apply(custom_filter)

STFT - Spectrogram Generation / STFT - スペクトログラム生成

Convert time-domain data to spectrograms: 時間領域データをスペクトログラムに変換：

# Create spectrogram dataset
# スペクトログラムデータセットを作成
spec_dataset = dataset.stft(
    n_fft=2048,
    hop_length=512,
    window="hann"
)

# Access a spectrogram
# スペクトログラムにアクセス
spec_frame = spec_dataset[0]
spec_frame.plot()

Iteration / 反復処理

Process all files in the dataset: データセット内のすべてのファイルを処理：

for i in range(len(dataset)):
    frame = dataset[i]
    if frame is not None:
        # Process the frame
        # フレームを処理
        print(f"Processing {frame.label}...")

Key Parameters / 主なパラメータ

folder_path (str): Path to the folder containing audio files. 音声ファイルを含むフォルダへのパス。

sampling_rate (Optional[int]): Target sampling rate. Files will be resampled if different from this rate. ターゲットサンプリングレート。このレートと異なる場合、ファイルはリサンプリングされます。

file_extensions (Optional[list[str]]): List of file extensions to include. By default, all registered reader extensions from wd.supported_formats() are used: [".aif", ".aiff", ".csv", ".flac", ".ogg", ".snd", ".wav"]. Audio format availability follows the libsndfile library bundled or installed with SoundFile. MP3 is not a registered Wandas reader format. 含めるファイル拡張子のリスト。デフォルトでは、wd.supported_formats() が返す登録済み reader 拡張子 [".aif", ".aiff", ".csv", ".flac", ".ogg", ".snd", ".wav"] をすべて使用します。音声形式の利用可否は SoundFile に同梱またはインストールされた libsndfile に従います。MP3 は Wandas の登録済み reader 形式ではありません。

lazy_loading (bool): If True, files are loaded only when accessed. Default: True. Trueの場合、ファイルはアクセス時にのみ読み込まれます。デフォルト: True。

recursive (bool): If True, search subdirectories recursively. Default: False. Trueの場合、サブディレクトリを再帰的に検索します。デフォルト: False。

path_metadata (bool): If True, infer AWS Glue-style metadata from relative parent folders. Prefer this option for ordinary partitioned folders. Default: False. Trueの場合、相対親フォルダからAWS Glue形式のメタデータを推論します。通常のパーティションフォルダではこの方法を優先します。デフォルト: False。

metadata_resolver (Callable[[Path], Mapping[str, object]] | None): Resolve file metadata from each root-relative path during discovery. Default: None. 探索時に各ルート相対パスからファイルメタデータを解決します。デフォルト: None。

Examples / 使用例

For detailed examples, see the learning-path/ directory and the tutorial marimo apps listed in the Tutorial section. 詳細な例については、learning-path/ ディレクトリとチュートリアル marimo アプリを参照してください。

API Reference / APIリファレンス

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 base dataset class for processing files in a folder. Includes lazy loading capability to efficiently handle large datasets. Subclasses handle specific frame types (ChannelFrame, SpectrogramFrame, etc.).

Source code in wandas/utils/frame_dataset.py

class FrameDataset(Generic[F], ABC):
    """
    Abstract base dataset class for processing files in a folder.
    Includes lazy loading capability to efficiently handle large datasets.
    Subclasses handle specific frame types (ChannelFrame, SpectrogramFrame, etc.).
    """

    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:
        """Ensure the frame at the given index is loaded."""
        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).

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

        Returns
        -------
        F | None
            The frame if found, otherwise None.

        Examples
        --------
        >>> frame = dataset.get_by_label("sample_1.wav")
        >>> if frame:
        ...     print(frame.label)
        """
        # Keep for backward compatibility: return the first match but emit
        # a DeprecationWarning recommending `get_all_by_label`.
        all_matches = self.get_all_by_label(label)
        if len(all_matches) > 0:
            warnings.warn(
                "get_by_label() returns the first matching frame and is deprecated; "
                "use get_all_by_label() to obtain all matches.",
                DeprecationWarning,
                stacklevel=2,
            )
            return all_matches[0]
        return None

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

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

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

        Returns
        -------
        F | None or list[F]
            If `key` is an int, returns the frame or None. If `key` is a str,
            returns a list of matching frames (may be empty).

        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]":
        """Apply a function to the entire dataset to create a new dataset."""
        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:
        """Save processed frames to files."""
        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]":
        """Get a sample from the dataset."""
        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]:
        """Get metadata for the dataset."""
        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).

Parameters

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

Returns

F | None The frame if found, otherwise None.

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

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

    Returns
    -------
    F | None
        The frame if found, otherwise None.

    Examples
    --------
    >>> frame = dataset.get_by_label("sample_1.wav")
    >>> if frame:
    ...     print(frame.label)
    """
    # Keep for backward compatibility: return the first match but emit
    # a DeprecationWarning recommending `get_all_by_label`.
    all_matches = self.get_all_by_label(label)
    if len(all_matches) > 0:
        warnings.warn(
            "get_by_label() returns the first matching frame and is deprecated; "
            "use get_all_by_label() to obtain all matches.",
            DeprecationWarning,
            stacklevel=2,
        )
        return all_matches[0]
    return None

get_all_by_label(label)

Get all frames matching the given label (filename).

Parameters

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.

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

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

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

Returns

F | None or list[F] If key is an int, returns the frame or None. If key is a str, returns a list of matching frames (may be empty).

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

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

    Returns
    -------
    F | None or list[F]
        If `key` is an int, returns the frame or None. If `key` is a str,
        returns a list of matching frames (may be empty).

    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]

Apply a function to the entire dataset to create a new dataset.

Source code in wandas/utils/frame_dataset.py

def apply(self, func: Callable[[F], Any | None]) -> "FrameDataset[Any]":
    """Apply a function to the entire dataset to create a new dataset."""
    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='')

Save processed frames to files.

Source code in wandas/utils/frame_dataset.py

def save(self, output_folder: str, filename_prefix: str = "") -> None:
    """Save processed frames to files."""
    raise NotImplementedError("The save method is not currently implemented.")

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

Get a sample from the dataset.

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]":
    """Get a sample from the dataset."""
    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()

Get metadata for the dataset.

Source code in wandas/utils/frame_dataset.py

def get_metadata(self) -> dict[str, Any]:
    """Get metadata for the dataset."""
    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

Sample Generation / サンプル生成

Provides functions for generating sample data for testing. テスト用のサンプルデータを生成する機能を提供します。

wandas.utils.generate_sample

Classes

Functions

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

Generate sample sine wave signals.

Parameters

freqs : float or list of float, default=1000 Frequency of the sine wave(s) in Hz. If multiple frequencies are specified, multiple channels will be created. 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 ChannelFrame object containing the sine wave(s).

Source code in wandas/utils/generate_sample.py

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

    Parameters
    ----------
    freqs : float or list of float, default=1000
        Frequency of the sine wave(s) in Hz.
        If multiple frequencies are specified, multiple channels will be created.
    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
        ChannelFrame object containing the sine wave(s).
    """
    # Directly call generate_sin_lazy function
    return generate_sin_lazy(freqs=freqs, sampling_rate=sampling_rate, duration=duration, label=label)

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

Generate sample sine wave signals using lazy computation.

Parameters

freqs : float or list of float, default=1000 Frequency of the sine wave(s) in Hz. If multiple frequencies are specified, multiple channels will be created. 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 Lazy ChannelFrame object containing the sine wave(s).

Source code in wandas/utils/generate_sample.py

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

    Parameters
    ----------
    freqs : float or list of float, default=1000
        Frequency of the sine wave(s) in Hz.
        If multiple frequencies are specified, multiple channels will be created.
    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
        Lazy ChannelFrame object containing the sine wave(s).
    """
    from wandas.frames.channel import ChannelFrame

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

    _freqs: list[float]
    if isinstance(freqs, float):
        _freqs = [freqs]
    elif isinstance(freqs, list):
        _freqs = freqs
    else:
        raise ValueError("freqs must be a float or a list of floats.")

    channels = []
    labels = []
    for idx, freq in enumerate(_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,
    )

Type Definitions / 型定義

Provides type definitions used in Wandas. Wandasで使用される型定義を提供します。

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

General Utilities / 一般ユーティリティ

Provides other general utility functions. その他の一般的なユーティリティ機能を提供します。

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

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

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.

    Parameters
    ----------
    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)
    level: NDArrayReal = 20 * np.log10(np.maximum(data / ref_shape, DB_FLOOR))
    return level

validate_sampling_rate(sampling_rate, param_name='sampling_rate')

Validate that sampling rate is positive.

Parameters

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

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.

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

unit : str Unit string.

Returns

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.

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

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.

Source code in wandas/utils/util.py

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

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

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.

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.

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

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

Returns

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.

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

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.

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.

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

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

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.

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