Dataset

Dataset and Corpus classes for managing collections of media files.

:class:MgDataset manages a collection of media files (video or audio) and provides batch processing, train/test splitting, and metadata management, following conventions from :mod:librosa and MNE-Python.

:class:MgCorpus is a higher-level convenience wrapper that scans a directory tree for media files and builds an :class:MgDataset automatically.

Examples

from musicalgestures._dataset import MgDataset ds = MgDataset.from_directory("/path/to/videos", pattern="*.avi") train, test = ds.train_test_split(test_size=0.2) for item in train: ... print(item["path"], item["label"])

MediaItem dataclass

MediaItem(path, label=None, metadata=dict())

A single item in an :class:MgDataset.

Parameters:

Name	Type	Description	Default
path	Path	Absolute path to the media file.	required
label	str | None	Optional class label or annotation string.	None
metadata	dict[str, Any]	Optional free-form metadata dict.	dict()

stem property

stem

Filename without extension.

suffix property

suffix

File extension (lower-case).

is_video property

is_video

True if this is a recognised video file.

is_audio property

is_audio

True if this is a recognised audio file.

MgDataset

MgDataset(items=None, name='MgDataset')

A labelled collection of media files.

Parameters:

Name	Type	Description	Default
items	list[MediaItem] | None	List of :class:MediaItem objects.	None
name	str	Optional human-readable name for this dataset.	'MgDataset'

Examples:

>>> from pathlib import Path
>>> from musicalgestures._dataset import MgDataset, MediaItem
>>> items = [
...     MediaItem(Path("/data/dance1.avi"), label="dance"),
...     MediaItem(Path("/data/piano1.avi"), label="piano"),
... ]
>>> ds = MgDataset(items, name="demo")
>>> len(ds)
2

Source code in musicalgestures/_dataset.py

def __init__(
    self,
    items: list[MediaItem] | None = None,
    name: str = "MgDataset",
) -> None:
    self._items: list[MediaItem] = list(items) if items else []
    self.name = name

labels property

labels

List of all item labels (in order).

unique_labels property

unique_labels

Sorted list of unique non-None labels.

from_directory classmethod

from_directory(directory, pattern='**/*', label_from='parent', recursive=True, name=None)

Build a dataset by scanning a directory for media files.

Parameters:

Name	Type	Description	Default
directory	str | Path	Root directory to scan.	required
pattern	str	Glob pattern relative to directory. Default: '**/*'.	'**/*'
label_from	str	How to derive labels: 'parent' uses the immediate parent directory name; 'stem' uses the filename stem; 'none' assigns no label.	'parent'
recursive	bool	If True (default), scan sub-directories.	True
name	str | None	Optional dataset name.	None

Returns:

Type	Description
MgDataset

Source code in musicalgestures/_dataset.py

@classmethod
def from_directory(
    cls,
    directory: str | Path,
    pattern: str = "**/*",
    label_from: str = "parent",
    recursive: bool = True,
    name: str | None = None,
) -> "MgDataset":
    """Build a dataset by scanning a directory for media files.

    Parameters
    ----------
    directory:
        Root directory to scan.
    pattern:
        Glob pattern relative to *directory*. Default: ``'**/*'``.
    label_from:
        How to derive labels: ``'parent'`` uses the immediate parent
        directory name; ``'stem'`` uses the filename stem;
        ``'none'`` assigns no label.
    recursive:
        If *True* (default), scan sub-directories.
    name:
        Optional dataset name.

    Returns
    -------
    MgDataset
    """
    root = Path(directory)
    if not root.is_dir():
        raise NotADirectoryError(f"Not a directory: {root}")

    all_extensions = _VIDEO_EXTENSIONS | _AUDIO_EXTENSIONS
    items: list[MediaItem] = []

    for p in sorted(root.glob(pattern)):
        if not p.is_file():
            continue
        if p.suffix.lower() not in all_extensions:
            continue
        label: str | None = None
        if label_from == "parent":
            label = p.parent.name
        elif label_from == "stem":
            label = p.stem
        items.append(MediaItem(path=p.resolve(), label=label))

    ds_name = name or root.name
    logger.info("Loaded %d media files from '%s'", len(items), root)
    return cls(items, name=ds_name)

from_json classmethod

from_json(path)

Load a dataset from a JSON file saved by :meth:to_json.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the JSON file.	required

Returns:

Type	Description
MgDataset

Source code in musicalgestures/_dataset.py

@classmethod
def from_json(cls, path: str | Path) -> "MgDataset":
    """Load a dataset from a JSON file saved by :meth:`to_json`.

    Parameters
    ----------
    path:
        Path to the JSON file.

    Returns
    -------
    MgDataset
    """
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    items = [
        MediaItem(
            path=Path(item["path"]),
            label=item.get("label"),
            metadata=item.get("metadata", {}),
        )
        for item in data["items"]
    ]
    return cls(items, name=data.get("name", "MgDataset"))

train_test_split

train_test_split(test_size=0.2, shuffle=True, seed=None)

Split the dataset into train and test subsets.

Parameters:

Name	Type	Description	Default
test_size	float	Fraction of items to include in the test set. Default: 0.2.	0.2
shuffle	bool	Whether to shuffle before splitting. Default: True.	True
seed	int | None	Random seed for reproducibility.	None

Returns:

Name	Type	Description
train	MgDataset
test	MgDataset

Source code in musicalgestures/_dataset.py

def train_test_split(
    self,
    test_size: float = 0.2,
    shuffle: bool = True,
    seed: int | None = None,
) -> tuple["MgDataset", "MgDataset"]:
    """Split the dataset into train and test subsets.

    Parameters
    ----------
    test_size:
        Fraction of items to include in the test set. Default: 0.2.
    shuffle:
        Whether to shuffle before splitting. Default: True.
    seed:
        Random seed for reproducibility.

    Returns
    -------
    train : MgDataset
    test  : MgDataset
    """
    items = list(self._items)
    if shuffle:
        rng = random.Random(seed)
        rng.shuffle(items)
    n_test = max(1, int(len(items) * test_size))
    test_items = items[:n_test]
    train_items = items[n_test:]
    return (
        MgDataset(train_items, name=f"{self.name}_train"),
        MgDataset(test_items, name=f"{self.name}_test"),
    )

filter

filter(func)

Return a new dataset containing only items for which func(item) is True.

Parameters:

Name	Type	Description	Default
func		Callable accepting a :class:MediaItem and returning bool.	required

Returns:

Type	Description
MgDataset

Source code in musicalgestures/_dataset.py

def filter(self, func) -> "MgDataset":
    """Return a new dataset containing only items for which *func(item)* is True.

    Parameters
    ----------
    func:
        Callable accepting a :class:`MediaItem` and returning bool.

    Returns
    -------
    MgDataset
    """
    return MgDataset([item for item in self._items if func(item)], name=self.name)

filter_by_label

filter_by_label(label)

Return a new dataset containing only items with the given label.

Parameters:

Name	Type	Description	Default
label	str	Label string to match.	required

Returns:

Type	Description
MgDataset

Source code in musicalgestures/_dataset.py

def filter_by_label(self, label: str) -> "MgDataset":
    """Return a new dataset containing only items with the given *label*.

    Parameters
    ----------
    label:
        Label string to match.

    Returns
    -------
    MgDataset
    """
    return self.filter(lambda item: item.label == label)

to_json

to_json(path=None)

Serialise the dataset to JSON.

Parameters:

Name	Type	Description	Default
path	str | Path | None	Optional file path to write. If None, returns the JSON string.	None

Returns:

Type	Description
str

Source code in musicalgestures/_dataset.py

def to_json(self, path: str | Path | None = None) -> str:
    """Serialise the dataset to JSON.

    Parameters
    ----------
    path:
        Optional file path to write.  If *None*, returns the JSON string.

    Returns
    -------
    str
    """
    payload = {
        "name": self.name,
        "n_items": len(self._items),
        "items": [
            {"path": str(item.path), "label": item.label, "metadata": item.metadata}
            for item in self._items
        ],
    }
    json_str = json.dumps(payload, indent=2)
    if path is not None:
        Path(path).write_text(json_str, encoding="utf-8")
        logger.info("MgDataset saved to %s", path)
    return json_str

MgCorpus

MgCorpus(root, pattern='**/*', label_from='parent')

Bases: MgDataset

Corpus: an :class:MgDataset built by scanning a directory tree.

This is a convenience subclass. Use :meth:MgDataset.from_directory for equivalent functionality.

Parameters:

Name	Type	Description	Default
root	str | Path	Root directory of the corpus.	required
pattern	str	Glob pattern. Default: '**/*'.	'**/*'
label_from	str	'parent', 'stem', or 'none'. Default: 'parent'.	'parent'

Examples:

>>> corpus = MgCorpus("/data/recordings", label_from="parent")
>>> len(corpus)
120
>>> train, test = corpus.train_test_split(test_size=0.2)

Source code in musicalgestures/_dataset.py

def __init__(
    self,
    root: str | Path,
    pattern: str = "**/*",
    label_from: str = "parent",
) -> None:
    ds = MgDataset.from_directory(root, pattern=pattern, label_from=label_from)
    super().__init__(ds._items, name=Path(root).name)
    self.root = Path(root)