Streaming

rddac.streaming is the offline-iteration namespace with four torch-free entry points:

• iter_view walks a Croissant view record by record. Shares a unified index that recognises loose .h5 files (rddac download --extract --remove-zip) and zipped *.zip archives interchangeably.
• export_to_numpy materialises a view as flat .npy memmap shards, with optional per-field and whole-record transforms. Requires every record to share the same shape per field.
• export_to_numpy_per_sim writes one .npz per experiment instead — the escape hatch for views whose raw fields have per-experiment sample counts.
• load_export opens the memmap shards back as a len + getitem + iter protocol object that plugs into torch.utils.data.DataLoader, tf.data.Dataset.from_generator, JAX, or plain Python without any adapter.

rddac.streaming.iter_view

iter_view(view: str, *, source: str | Path | None = None, data_dir: str | Path | None = DEFAULT_DATA_DIR, dataset=None, sim_ids: list[int] | None = None, where: Callable[[pd.Series], bool] | None = None) -> Iterator[dict[str, np.ndarray]]

Yield one record per RDDAC experiment for a Croissant view.

Parameters:

Name	Type	Description	Default
view	str	Name of the RecordSet to stream (published, e.g. force-curve, or added via :func:rddac.add_view).	required
source	str | Path | None	Override the Croissant manifest URL/path.	None
data_dir	str | Path | None	Directory holding metadata.json, process_parameters.csv and either loose h5/<id>.h5 files or the dataset zips.	DEFAULT_DATA_DIR
dataset		A pre-loaded mlcroissant.Dataset (carries add_view mutations).	None
sim_ids	list[int] | None	Optional allowlist of experiment ids (name kept for drop-in DDACS compatibility). Requested ids that cannot be served warn via :class:MissingDataWarning.	None
where	Callable[[Series], bool] | None	Predicate applied to each process_parameters.csv row before any HDF5 file is touched.	None

Yields:

Type	Description
dict[str, ndarray]	A dict[str, np.ndarray] per experiment, keyed by view-field aliases
dict[str, ndarray]	(plus the private _sim_id scratch key).

Source code in rddac/streaming.py

def iter_view(
    view: str,
    *,
    source: str | Path | None = None,
    data_dir: str | Path | None = DEFAULT_DATA_DIR,
    dataset=None,
    sim_ids: list[int] | None = None,
    where: Callable[[pd.Series], bool] | None = None,
) -> Iterator[dict[str, np.ndarray]]:
    """Yield one record per RDDAC experiment for a Croissant view.

    Args:
        view: Name of the RecordSet to stream (published, e.g. ``force-curve``,
            or added via :func:`rddac.add_view`).
        source: Override the Croissant manifest URL/path.
        data_dir: Directory holding ``metadata.json``, ``process_parameters.csv``
            and either loose ``h5/<id>.h5`` files or the dataset zips.
        dataset: A pre-loaded ``mlcroissant.Dataset`` (carries ``add_view``
            mutations).
        sim_ids: Optional allowlist of experiment ids (name kept for drop-in
            DDACS compatibility). Requested ids that cannot be served warn via
            :class:`MissingDataWarning`.
        where: Predicate applied to each ``process_parameters.csv`` row before
            any HDF5 file is touched.

    Yields:
        A ``dict[str, np.ndarray]`` per experiment, keyed by view-field aliases
        (plus the private ``_sim_id`` scratch key).
    """
    return _ddacs_streaming.iter_view(
        view,
        source=source,
        data_dir=data_dir,
        dataset=dataset,
        sim_ids=sim_ids,
        where=where,
        spec=RDDAC_SPEC,
    )

rddac.streaming.export_to_numpy

export_to_numpy(view: str, out_dir: str | Path, *, source: str | Path | None = None, data_dir: str | Path | None = DEFAULT_DATA_DIR, dataset=None, sim_ids: list[int] | None = None, where: Callable[[pd.Series], bool] | None = None, transforms: dict[str, Callable[[Any], Any]] | None = None, record_transform: Callable[[dict[str, Any]], dict[str, Any]] | None = None, show_progress: bool = False) -> dict[str, Path]

Materialize a Croissant view as flat .npy memmaps on disk.

Requires every record to share the same shape per field — raw RDDAC force and traverse tables vary per experiment, so either slice/resample them via record_transform or use :func:export_to_numpy_per_sim. See :func:ddacs.streaming.export_to_numpy for the full parameter reference.

Source code in rddac/streaming.py

def export_to_numpy(
    view: str,
    out_dir: str | Path,
    *,
    source: str | Path | None = None,
    data_dir: str | Path | None = DEFAULT_DATA_DIR,
    dataset=None,
    sim_ids: list[int] | None = None,
    where: Callable[[pd.Series], bool] | None = None,
    transforms: dict[str, Callable[[Any], Any]] | None = None,
    record_transform: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
    show_progress: bool = False,
) -> dict[str, Path]:
    """Materialize a Croissant view as flat ``.npy`` memmaps on disk.

    Requires every record to share the same shape per field — raw RDDAC force
    and traverse tables vary per experiment, so either slice/resample them via
    ``record_transform`` or use :func:`export_to_numpy_per_sim`. See
    :func:`ddacs.streaming.export_to_numpy` for the full parameter reference.
    """
    return _ddacs_streaming.export_to_numpy(
        view,
        out_dir,
        source=source,
        data_dir=data_dir,
        dataset=dataset,
        sim_ids=sim_ids,
        where=where,
        transforms=transforms,
        record_transform=record_transform,
        show_progress=show_progress,
        spec=RDDAC_SPEC,
    )

rddac.streaming.export_to_numpy_per_sim

export_to_numpy_per_sim(view: str, out_dir: str | Path, *, source: str | Path | None = None, data_dir: str | Path | None = DEFAULT_DATA_DIR, dataset=None, sim_ids: list[int] | None = None, where: Callable[[pd.Series], bool] | None = None, transforms: dict[str, Callable[[Any], Any]] | None = None, record_transform: Callable[[dict[str, Any]], dict[str, Any]] | None = None, compressed: bool = False, show_progress: bool = False) -> Path

Write one <experiment_id>.npz per experiment under out_dir.

Same pipeline as :func:export_to_numpy but fields may have experiment-dependent shapes (the natural fit for RDDAC's raw tables). See :func:ddacs.streaming.export_to_numpy_per_sim for details.

Source code in rddac/streaming.py

def export_to_numpy_per_sim(
    view: str,
    out_dir: str | Path,
    *,
    source: str | Path | None = None,
    data_dir: str | Path | None = DEFAULT_DATA_DIR,
    dataset=None,
    sim_ids: list[int] | None = None,
    where: Callable[[pd.Series], bool] | None = None,
    transforms: dict[str, Callable[[Any], Any]] | None = None,
    record_transform: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
    compressed: bool = False,
    show_progress: bool = False,
) -> Path:
    """Write one ``<experiment_id>.npz`` per experiment under ``out_dir``.

    Same pipeline as :func:`export_to_numpy` but fields may have
    experiment-dependent shapes (the natural fit for RDDAC's raw tables). See
    :func:`ddacs.streaming.export_to_numpy_per_sim` for details.
    """
    return _ddacs_streaming.export_to_numpy_per_sim(
        view,
        out_dir,
        source=source,
        data_dir=data_dir,
        dataset=dataset,
        sim_ids=sim_ids,
        where=where,
        transforms=transforms,
        record_transform=record_transform,
        compressed=compressed,
        show_progress=show_progress,
        spec=RDDAC_SPEC,
    )

rddac.streaming.load_export

load_export(directory: str | Path, fields: list[str] | None = None) -> _LoadedExport

Open a directory of .npy shards produced by :func:export_to_numpy.

Returns a lazy reader that exposes the standard Python data model (len(reader), reader[i], for r in reader, reader.by_sim_id(sid)). Each record is a plain dict[str, numpy.ndarray] backed by mmap_mode='r', so the full release fits even when it exceeds RAM -- only the rows actually accessed page in.

The returned object is framework-agnostic. Pass it directly into torch.utils.data.DataLoader(loader, batch_size=...) because it satisfies the map-style Dataset protocol. The same instance also works with tf.data.Dataset.from_generator(lambda: iter(loader), ...), with JAX, or with plain Python loops -- no adapter required and no torch import inside.

Parameters:

Name	Type	Description	Default
directory	str | Path	Path produced by :func:export_to_numpy. Must contain a sim_ids.npy and at least one other .npy shard.	required
fields	list[str] | None	Optional subset of field aliases to load. None loads every shard in the directory. Unknown names raise ValueError so typos surface immediately, listing the available fields.	None

Returns:

Type	Description
_LoadedExport	A reader instance. The concrete type is private; rely on the
_LoadedExport	documented protocol rather than the class name.

Raises:

Type	Description
FileNotFoundError	directory is empty or missing sim_ids.npy.
ValueError	fields references a name that is not on disk.

Example

export = ddacs.streaming.load_export("./data/tutorial_export") len(export), export.fields # (1, ('delta', 'forming', ...)) export[0] # {alias: ndarray} export.by_sim_id(258864) # lookup by simulation id from torch.utils.data import DataLoader DataLoader(export, batch_size=16, shuffle=True)

Source code in ddacs/streaming.py

def load_export(directory: str | Path, fields: list[str] | None = None) -> _LoadedExport:
    """Open a directory of ``.npy`` shards produced by :func:`export_to_numpy`.

    Returns a lazy reader that exposes the standard Python data model
    (``len(reader)``, ``reader[i]``, ``for r in reader``, ``reader.by_sim_id(sid)``).
    Each record is a plain ``dict[str, numpy.ndarray]`` backed by
    ``mmap_mode='r'``, so the full release fits even when it exceeds RAM --
    only the rows actually accessed page in.

    The returned object is framework-agnostic. Pass it directly into
    ``torch.utils.data.DataLoader(loader, batch_size=...)`` because it
    satisfies the map-style ``Dataset`` protocol. The same instance also
    works with ``tf.data.Dataset.from_generator(lambda: iter(loader), ...)``,
    with JAX, or with plain Python loops -- no adapter required and no
    ``torch`` import inside.

    Args:
        directory: Path produced by :func:`export_to_numpy`. Must contain a
            ``sim_ids.npy`` and at least one other ``.npy`` shard.
        fields: Optional subset of field aliases to load. ``None`` loads
            every shard in the directory. Unknown names raise ``ValueError``
            so typos surface immediately, listing the available fields.

    Returns:
        A reader instance. The concrete type is private; rely on the
        documented protocol rather than the class name.

    Raises:
        FileNotFoundError: ``directory`` is empty or missing ``sim_ids.npy``.
        ValueError: ``fields`` references a name that is not on disk.

    Example:
        >>> export = ddacs.streaming.load_export("./data/tutorial_export")
        >>> len(export), export.fields                # (1, ('delta', 'forming', ...))
        >>> export[0]                                 # {alias: ndarray}
        >>> export.by_sim_id(258864)                  # lookup by simulation id
        >>> from torch.utils.data import DataLoader
        >>> DataLoader(export, batch_size=16, shuffle=True)
    """
    return _LoadedExport(directory, fields=fields)