Streaming

ddacs.streaming is the offline-iteration namespace with three torch-free entry points:

• iter_view walks a Croissant view record by record. Shares a unified index that recognises loose .h5 files (ddacs 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.
• load_export opens those 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.

ddacs.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, spec: DatasetSpec = DDACS_SPEC) -> Iterator[dict[str, np.ndarray]]

Yield one record per simulation for a Croissant view.

Parameters:

Name	Type	Description	Default
view	str	Name of the RecordSet to stream. Must exist on the loaded mlcroissant.Dataset (either as a published RecordSet or a view added via :func:ddacs.add_view).	required
source	str | Path | None	Override the Croissant manifest URL/path. Ignored when dataset= is given.	None
data_dir	str | Path | None	Directory holding metadata.json, process_parameters.csv and either loose h5/<sim_id>.h5 files or zipped h5/*.zip archives.	DEFAULT_DATA_DIR
dataset		A pre-loaded mlcroissant.Dataset (e.g. one mutated by :func:ddacs.add_view). When supplied, the manifest is not re-parsed and the caller's object is used as-is.	None
sim_ids	list[int] | None	Optional allowlist of simulation ids; iteration order is preserved from the CSV when sim_ids is not supplied, otherwise from the allowlist.	None
where	Callable[[Series], bool] | None	Predicate applied to each process_parameters.csv row before any HDF5 file is touched. Combined with sim_ids if both are given.	None

Yields:

Type	Description
dict[str, ndarray]	A dict[str, np.ndarray] per simulation, keyed by the view-field
dict[str, ndarray]	aliases. Sliced fields come out with the requested timesteps
dict[str, ndarray]	applied; whole fields come out at their full shape.

Source code in ddacs/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,
    spec: DatasetSpec = DDACS_SPEC,
) -> Iterator[dict[str, np.ndarray]]:
    """Yield one record per simulation for a Croissant view.

    Args:
        view: Name of the RecordSet to stream. Must exist on the loaded
            ``mlcroissant.Dataset`` (either as a published RecordSet or a
            view added via :func:`ddacs.add_view`).
        source: Override the Croissant manifest URL/path. Ignored when
            ``dataset=`` is given.
        data_dir: Directory holding ``metadata.json``,
            ``process_parameters.csv`` and either loose ``h5/<sim_id>.h5``
            files or zipped ``h5/*.zip`` archives.
        dataset: A pre-loaded ``mlcroissant.Dataset`` (e.g. one mutated by
            :func:`ddacs.add_view`). When supplied, the manifest is not
            re-parsed and the caller's object is used as-is.
        sim_ids: Optional allowlist of simulation ids; iteration order is
            preserved from the CSV when ``sim_ids`` is not supplied,
            otherwise from the allowlist.
        where: Predicate applied to each ``process_parameters.csv`` row
            before any HDF5 file is touched. Combined with ``sim_ids`` if
            both are given.

    Yields:
        A ``dict[str, np.ndarray]`` per simulation, keyed by the view-field
        aliases. Sliced fields come out with the requested timesteps
        applied; whole fields come out at their full shape.
    """
    ds = (
        dataset
        if dataset is not None
        else _croissant.load(source=source, data_dir=data_dir, spec=spec)
    )
    h5_specs, csv_specs = _build_field_specs(ds, view, spec)
    h5_index = _build_unified_index(Path(data_dir) if data_dir is not None else None)
    final_ids = _resolve_sim_ids(
        Path(data_dir) if data_dir is not None else None,
        h5_index,
        sim_ids,
        where,
        spec,
    )

    csv_df: pd.DataFrame | None = None
    if csv_specs:
        if data_dir is None:
            raise ValueError(f"view {view!r} pulls from 'process-parameters'; pass data_dir=")
        csv_path = Path(data_dir) / spec.process_parameters_file
        csv_df = pd.read_csv(csv_path).set_index(spec.id_column)

    if sim_ids is not None:
        _warn_missing(sim_ids, final_ids, h5_index, data_dir)

    last_zip_path: str | None = None
    last_zf: zipfile.ZipFile | None = None
    try:
        for sim_id in final_ids:
            path = h5_index.get(int(sim_id))
            if path is None:
                continue

            if path.endswith(".h5"):
                # Loose file: open directly, no zip indirection.
                with h5py.File(path, "r") as f:
                    record = _extract_record(f, h5_specs)
            elif path.endswith(".zip"):
                # Cache the zip handle across consecutive sims that land in
                # the same archive; corner-block zips group ~2200 sims each.
                if path != last_zip_path:
                    if last_zf is not None:
                        last_zf.close()
                    last_zf = zipfile.ZipFile(path)
                    last_zip_path = path
                try:
                    data = last_zf.read(f"{spec.id_format.format(int(sim_id))}.h5")
                except KeyError:
                    continue
                with h5py.File(io.BytesIO(data), "r") as f:
                    record = _extract_record(f, h5_specs)
            else:
                continue
            if csv_df is not None:
                row = csv_df.loc[int(sim_id)]
                for alias, col in csv_specs.items():
                    record[alias] = row[col]
            # Private scratch key for transforms (e.g. per-sim RNG seeding).
            # export_to_numpy strips it before writing memmaps.
            record["_sim_id"] = int(sim_id)
            yield record
    finally:
        if last_zf is not None:
            last_zf.close()

ddacs.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, spec: DatasetSpec = DDACS_SPEC) -> dict[str, Path]

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

Iterates the view once (via :func:iter_view), applies optional per-field transforms, and writes each output alias into its own pre-allocated numpy.memmap of shape (n_sims, *field_shape). Also writes sim_ids.npy so the row order is recoverable.

After this runs once, the per-record cost drops from "open zip + h5py + decompress" to a single mmap'd numpy slice, which is the lever the Croissant + HDF5 layer cannot match on its own.

Parameters:

Name	Type	Description	Default
view	str	Name of the RecordSet to export.	required
out_dir	str | Path	Output directory. Created if missing.	required
source	str | Path | None	Override the manifest URL/path. Ignored when dataset= is given.	None
data_dir	str | Path | None	Local data directory used to discover the simulations.	DEFAULT_DATA_DIR
dataset		Pre-loaded mlcroissant.Dataset (carries any add_view mutations).	None
sim_ids	list[int] | None	Optional allowlist of simulation ids.	None
where	Callable[[Series], bool] | None	Optional predicate on process_parameters.csv rows.	None
transforms	dict[str, Callable[[Any], Any]] | None	Per-field encoder callable, {alias: fn(raw_value)}. Applied before the record is written. Use for non-numeric columns (e.g. "geometry" -> integer label) or for dtype downcasts. Identity is assumed for any alias not listed.	None
record_transform	Callable[[dict[str, Any]], dict[str, Any]] | None	Optional full-record callable, runs after the per-field transforms. Can add, drop or rename fields. The returned dict determines which .npy files end up on disk.	None
show_progress	bool	If True and tqdm is importable, display a progress bar.	False

Returns:

Type	Description
dict[str, Path]	A dict {alias: Path} mapping each output field's name to its
dict[str, Path]	.npy file. sim_ids is included under the same name.

Raises:

Type	Description
ValueError	If the view yields no records, or if a transformed value cannot be coerced to a numpy array.

Source code in ddacs/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,
    spec: DatasetSpec = DDACS_SPEC,
) -> dict[str, Path]:
    """Materialize a Croissant view as flat ``.npy`` files on disk.

    Iterates the view once (via :func:`iter_view`), applies optional
    per-field transforms, and writes each output alias into its own
    pre-allocated ``numpy.memmap`` of shape ``(n_sims, *field_shape)``.
    Also writes ``sim_ids.npy`` so the row order is recoverable.

    After this runs once, the per-record cost drops from "open zip + h5py
    + decompress" to a single mmap'd numpy slice, which is the lever the
    Croissant + HDF5 layer cannot match on its own.

    Args:
        view: Name of the RecordSet to export.
        out_dir: Output directory. Created if missing.
        source: Override the manifest URL/path. Ignored when ``dataset=`` is
            given.
        data_dir: Local data directory used to discover the simulations.
        dataset: Pre-loaded ``mlcroissant.Dataset`` (carries any ``add_view``
            mutations).
        sim_ids: Optional allowlist of simulation ids.
        where: Optional predicate on ``process_parameters.csv`` rows.
        transforms: Per-field encoder callable, ``{alias: fn(raw_value)}``.
            Applied before the record is written. Use for non-numeric columns
            (e.g. ``"geometry"`` -> integer label) or for dtype downcasts.
            Identity is assumed for any alias not listed.
        record_transform: Optional full-record callable, runs after the
            per-field transforms. Can add, drop or rename fields. The
            returned dict determines which ``.npy`` files end up on disk.
        show_progress: If ``True`` and ``tqdm`` is importable, display a
            progress bar.

    Returns:
        A dict ``{alias: Path}`` mapping each output field's name to its
        ``.npy`` file. ``sim_ids`` is included under the same name.

    Raises:
        ValueError: If the view yields no records, or if a transformed
            value cannot be coerced to a numpy array.
    """
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    transforms = transforms or {}

    # Build the manifest + index once so iter_view doesn't redo the work.
    ds = (
        dataset
        if dataset is not None
        else _croissant.load(source=source, data_dir=data_dir, spec=spec)
    )

    records = iter_view(
        view=view,
        source=source,
        data_dir=data_dir,
        dataset=ds,
        sim_ids=sim_ids,
        where=where,
        spec=spec,
    )

    # Pre-resolve the sim id order so we can both size the memmaps and
    # record which row of each output array corresponds to which sim.
    final_ids = _resolve_sim_ids(
        Path(data_dir) if data_dir is not None else None,
        _build_unified_index(Path(data_dir) if data_dir is not None else None),
        sim_ids,
        where,
        spec,
    )
    if not final_ids:
        raise ValueError("no simulations matched; nothing to export")

    progress = _progress_iter(records, total=len(final_ids), enabled=show_progress)

    try:
        first = next(progress)
    except StopIteration as exc:
        raise ValueError("view yielded no records") from exc

    first = _apply_transforms(first, transforms, record_transform)
    first.pop("_sim_id", None)
    if not first:
        raise ValueError("record_transform returned an empty dict; nothing to write")

    memmaps: dict[str, np.memmap] = {}
    paths: dict[str, Path] = {}
    for alias, val in first.items():
        arr = _as_array(alias, val)
        path = out_dir / f"{alias}.npy"
        mm = np.lib.format.open_memmap(
            path, mode="w+", dtype=arr.dtype, shape=(len(final_ids), *arr.shape)
        )
        mm[0] = arr
        memmaps[alias] = mm
        paths[alias] = path

    # Stream the rest.
    for i, rec in enumerate(progress, start=1):
        rec = _apply_transforms(rec, transforms, record_transform)
        rec.pop("_sim_id", None)
        for alias, val in rec.items():
            if alias not in memmaps:
                raise ValueError(
                    f"record {i} produced new alias {alias!r} not present in record 0; "
                    "all records must share the same output keys"
                )
            arr = _as_array(alias, val)
            expected_shape = memmaps[alias].shape[1:]
            if arr.shape != expected_shape:
                raise ValueError(
                    f"record {i}: field {alias!r} has shape {arr.shape} but record 0 "
                    f"had shape {expected_shape}. export_to_numpy requires every record "
                    f"to share the same shape per field (it pre-allocates one memmap per "
                    f"alias from record 0). For variable-shape views (e.g. graph data "
                    f"with sim-dependent node or edge counts) use "
                    f"ddacs.streaming.export_to_numpy_per_sim instead, which writes one "
                    f".npz per simulation and allows shapes to vary."
                )
            memmaps[alias][i] = arr

    for mm in memmaps.values():
        mm.flush()

    sim_ids_path = out_dir / "sim_ids.npy"
    np.save(sim_ids_path, np.asarray(final_ids, dtype=np.int64))
    paths["sim_ids"] = sim_ids_path

    return paths

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