# helia_edge.data.grain_data

Grain pipelines over a ``DataSource``.

## helia_edge.data.grain_data.to_grain

`function` · `python`

```python
to_grain(
    source: DataSource[Any],
    *,
    seed: int | None = None,
    shuffle: bool = False,
    transform: Callable[[Any, np.random.Generator], Any] | None = None,
    batch_size: int | None = None,
    drop_remainder: bool = False,
    num_epochs: int | None = 1,
    workers: int = 0,
    worker_buffer_size: int = 1,
) -> grain.IterDataset
```

Read ``source`` through Grain: shuffle, repeat, transform, batch and prefetch.

The order is source, shuffle (a new permutation each epoch), repeat, ``transform``, batch, so
batches run across epoch boundaries. Record order and transform randomness depend only on
``seed``, not on ``workers``. Every ``iter()`` replays the same elements, so a Keras ``fit``
that iterates once per epoch sees the same epoch each time: for Keras training, pass
``num_epochs=None`` and set ``steps_per_epoch``.

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| source | DataSource[Any] | Required | Records addressed by index. |
| seed | int \| None | None | Seed for the shuffle and for the per-record generator passed to ``transform``, from 0 to 2**32 - 1. Required when ``shuffle`` is set or ``transform`` is given. |
| shuffle | bool | False | Shuffle the records of every epoch (each of the ``num_epochs`` passes). |
| transform | Callable[[Any, np.random.Generator], Any] \| None | None | ``transform(record, rng) -> record``, run per record in the Grain workers; ``rng`` is a NumPy generator derived from ``seed`` and the record's position. |
| batch_size | int \| None | None | Stack this many records into one batch of NumPy arrays; None keeps single records. |
| drop_remainder | bool | False | Drop the last batch if it is smaller than ``batch_size``. |
| num_epochs | int \| None | 1 | Number of passes over ``source``; None repeats indefinitely. |
| workers | int | 0 | Grain worker processes; 0 reads in this process. |
| worker_buffer_size | int | 1 | Elements each worker prepares ahead. |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | grain.IterDataset | grain.IterDataset: Re-iterable; each ``iter()`` starts from the first element. |

**Raises**

| Name | Description |
| --- | --- |
| ValueError | If ``seed`` is missing when needed or out of range, or a count is not a positive integer (``workers`` may be 0). |
| ImportError | If Grain is not installed (``helia-edge[grain]``). |

Source: `helia_edge/data/grain_data.py:24`