7. User-defined datasets & transforms¶

Two registries let you plug your own code into the same dataset_from_config / YAML machinery used by official datasets:

• register_dataset (decorator) → adds a Dataset subclass
• register_transform(config_cls, transform_cls) → adds a transform

Setup: write a small synthetic CSV to disk¶

In [ ]:

import numpy as np
import pandas as pd

from alp_data.io import anypath, filesystem_from_path

# `anypath` returns a pathlib.Path for local paths (and PureGSPath / PureR2Path for
# cloud URIs), so the same `/` joining works everywhere. We grab the matching fsspec
# filesystem via `filesystem_from_path` to create the directory.
tmp = anypath("./_demo_data")
filesystem_from_path(tmp).makedirs(str(tmp), exist_ok=True)

rng = np.random.default_rng(0)
df = pd.DataFrame({
    "id": range(20),
    "species": rng.choice(["finch", "sparrow", "robin"], size=20),
    "duration": rng.uniform(0.5, 3.0, size=20).round(3),
})
df.to_csv(tmp / "train.csv", index=False)
df.head()

import numpy as np import pandas as pd from alp_data.io import anypath, filesystem_from_path # `anypath` returns a pathlib.Path for local paths (and PureGSPath / PureR2Path for # cloud URIs), so the same `/` joining works everywhere. We grab the matching fsspec # filesystem via `filesystem_from_path` to create the directory. tmp = anypath("./_demo_data") filesystem_from_path(tmp).makedirs(str(tmp), exist_ok=True) rng = np.random.default_rng(0) df = pd.DataFrame({ "id": range(20), "species": rng.choice(["finch", "sparrow", "robin"], size=20), "duration": rng.uniform(0.5, 3.0, size=20).round(3), }) df.to_csv(tmp / "train.csv", index=False) df.head()

Register a custom dataset¶

Subclass Dataset, define an info, implement the abstract methods, and decorate with @register_dataset.

In [2]:

from typing import Any, Iterator
import numpy as np

from alp_data import Dataset, DatasetInfo, DatasetConfig, register_dataset


@register_dataset
class TinyBirds(Dataset):
    """Synthetic in-memory bird dataset for demo purposes."""

    info = DatasetInfo(
        name="tiny_birds",
        owner="demo",
        split_paths={"train": str(tmp / "train.csv")},
        version="0.1.0",
        description="Tiny synthetic dataset used in the conference demo.",
        sources=["synthetic"],
        license="CC0",
    )

    def __init__(self, split: str = "train", output_take_and_give=None, backend="pandas", streaming=False):
        super().__init__(output_take_and_give, backend=backend, streaming=streaming)
        self.split = split
        self._data = None
        self._load()

    def _load(self) -> None:
        if self.split not in self.info.split_paths:
            raise LookupError(f"unknown split {self.split!r}")
        self._data = self._backend_class.from_csv(self.info.split_paths[self.split])

    @property
    def columns(self):
        return list(self._data.columns)

    @property
    def available_splits(self):
        return list(self.info.split_paths)

    def __len__(self) -> int:
        return len(self._data)

    def __getitem__(self, idx: int) -> dict[str, Any]:
        row = self._data[idx]
        # Fake audio so the sample shape matches what model code expects.
        n_samples = int(row["duration"] * 16000)
        row["audio"] = np.random.randn(n_samples).astype(np.float32)
        if self.output_take_and_give:
            return {dst: row[src] for src, dst in self.output_take_and_give.items()}
        return row

    def __iter__(self) -> Iterator[dict[str, Any]]:
        for i in range(len(self)):
            yield self[i]

    def __str__(self) -> str:
        return f"TinyBirds(split={self.split!r}, n={len(self)})"

    @classmethod
    def from_config(cls, cfg: DatasetConfig):
        ds = cls(split=cfg.split, output_take_and_give=cfg.output_take_and_give)
        meta = ds.apply_transformations(cfg.transformations) if cfg.transformations else {}
        return ds, meta

from typing import Any, Iterator import numpy as np from alp_data import Dataset, DatasetInfo, DatasetConfig, register_dataset @register_dataset class TinyBirds(Dataset): """Synthetic in-memory bird dataset for demo purposes.""" info = DatasetInfo( name="tiny_birds", owner="demo", split_paths={"train": str(tmp / "train.csv")}, version="0.1.0", description="Tiny synthetic dataset used in the conference demo.", sources=["synthetic"], license="CC0", ) def __init__(self, split: str = "train", output_take_and_give=None, backend="pandas", streaming=False): super().__init__(output_take_and_give, backend=backend, streaming=streaming) self.split = split self._data = None self._load() def _load(self) -> None: if self.split not in self.info.split_paths: raise LookupError(f"unknown split {self.split!r}") self._data = self._backend_class.from_csv(self.info.split_paths[self.split]) @property def columns(self): return list(self._data.columns) @property def available_splits(self): return list(self.info.split_paths) def __len__(self) -> int: return len(self._data) def __getitem__(self, idx: int) -> dict[str, Any]: row = self._data[idx] # Fake audio so the sample shape matches what model code expects. n_samples = int(row["duration"] * 16000) row["audio"] = np.random.randn(n_samples).astype(np.float32) if self.output_take_and_give: return {dst: row[src] for src, dst in self.output_take_and_give.items()} return row def __iter__(self) -> Iterator[dict[str, Any]]: for i in range(len(self)): yield self[i] def __str__(self) -> str: return f"TinyBirds(split={self.split!r}, n={len(self)})" @classmethod def from_config(cls, cfg: DatasetConfig): ds = cls(split=cfg.split, output_take_and_give=cfg.output_take_and_give) meta = ds.apply_transformations(cfg.transformations) if cfg.transformations else {} return ds, meta

In [3]:

from alp_data import list_registered_datasets, dataset_class_from_name

print("tiny_birds" in list_registered_datasets())
ds = dataset_class_from_name("tiny_birds")(split="train")
print(ds)
print("sample audio shape:", ds[0]["audio"].shape)

from alp_data import list_registered_datasets, dataset_class_from_name print("tiny_birds" in list_registered_datasets()) ds = dataset_class_from_name("tiny_birds")(split="train") print(ds) print("sample audio shape:", ds[0]["audio"].shape)

True
TinyBirds(split='train', n=20)
sample audio shape: (40640,)

Register a custom transform¶

A transform is (config_class, transform_class). The config class is a pydantic model with a Literal["<unique_type>"] type field. The transform class needs from_config and __call__(backend) -> (backend, metadata).

In [4]:

from typing import Literal
from pydantic import BaseModel
from alp_data.backends.protocol import DataBackend
from alp_data.transforms import register_transform


class TagSourceConfig(BaseModel):
    type: Literal["tag_source"]
    source_name: str


class TagSource:
    """Add a constant `source` column to every row."""

    def __init__(self, *, source_name: str) -> None:
        self.source_name = source_name

    @classmethod
    def from_config(cls, cfg: TagSourceConfig) -> "TagSource":
        return cls(source_name=cfg.source_name)

    def __call__(self, backend: DataBackend) -> tuple[DataBackend, dict]:
        new_backend = backend.add_column("source", [self.source_name] * len(backend))
        return new_backend, {"tagged_n": len(backend)}


register_transform(TagSourceConfig, TagSource)

from typing import Literal from pydantic import BaseModel from alp_data.backends.protocol import DataBackend from alp_data.transforms import register_transform class TagSourceConfig(BaseModel): type: Literal["tag_source"] source_name: str class TagSource: """Add a constant `source` column to every row.""" def __init__(self, *, source_name: str) -> None: self.source_name = source_name @classmethod def from_config(cls, cfg: TagSourceConfig) -> "TagSource": return cls(source_name=cfg.source_name) def __call__(self, backend: DataBackend) -> tuple[DataBackend, dict]: new_backend = backend.add_column("source", [self.source_name] * len(backend)) return new_backend, {"tagged_n": len(backend)} register_transform(TagSourceConfig, TagSource)

Use both together¶

In [5]:

from alp_data.transforms import LabelFromFeatureConfig

ds = dataset_class_from_name("tiny_birds")(split="train")
print("columns before:", ds.columns)

meta = ds.apply_transformations([
    TagSourceConfig(type="tag_source", source_name="demo"),
    LabelFromFeatureConfig(type="label_from_feature", feature="species", output_feature="label"),
])

print("columns after: ", ds.columns)
print("metadata:", meta)
print("first row keys:", list(ds[0].keys()))

from alp_data.transforms import LabelFromFeatureConfig ds = dataset_class_from_name("tiny_birds")(split="train") print("columns before:", ds.columns) meta = ds.apply_transformations([ TagSourceConfig(type="tag_source", source_name="demo"), LabelFromFeatureConfig(type="label_from_feature", feature="species", output_feature="label"), ]) print("columns after: ", ds.columns) print("metadata:", meta) print("first row keys:", list(ds[0].keys()))

columns before: ['id', 'species', 'duration']
columns after:  ['id', 'species', 'duration', 'source', 'label']
metadata: {'tag_source': {'tagged_n': 20}, 'label_from_feature': {'label_feature': 'species', 'label_map': {'finch': 0, 'robin': 1, 'sparrow': 2}, 'num_classes': 3}}
first row keys: ['id', 'species', 'duration', 'source', 'label', 'audio']

Both registries are global — once registered, the new dataset and transform also work through dataset_from_config and YAML, exactly like the built-ins.