# helia_edge.data.tf_data

TensorFlow ``tf.data`` adapters: generators, arrays and batches from other loaders.

## helia_edge.data.tf_data.convert_inputs_to_tf_dataset

`function` · `python`

```python
convert_inputs_to_tf_dataset(x=None, y=None, sample_weight=None, batch_size=None)
```

Convert inputs to tf.data.Dataset.

Source: `helia_edge/data/tf_data.py:18`

## helia_edge.data.tf_data.create_interleaved_dataset_from_generator

`function` · `python`

```python
create_interleaved_dataset_from_generator(
    data_generator: Callable[[Iterator[T]], Iterable[K]],
    id_generator: Callable[[list[T]], Iterator[T]],
    ids: list[T],
    spec: tf.TensorSpec | tuple[tf.TensorSpec, ...] | dict[str, tf.TensorSpec],
    preprocess: Callable[[K], K] | None = None,
    num_workers: int = 4,
    *,
    stream_mode: StreamMode | str = StreamMode.GLOBAL,
    deterministic: bool = True,
) -> tf.data.Dataset
```

Adapt caller-owned schedules to tf.data without changing sample weights.

GLOBAL preserves one finite/repeated stream. FINITE partitions terminating,
partition-independent generators; deterministic mode preserves partition order.
num_workers counts generators, not processes. See https://ambiqai.github.io/helia-edge/guide/input-pipeline/.

Source: `helia_edge/data/tf_data.py:56`

## helia_edge.data.tf_data.create_dataset_from_data

`function` · `python`

```python
create_dataset_from_data(x: npt.NDArray, y: npt.NDArray, spec: tuple[tf.TensorSpec, ...]) -> tf.data.Dataset
```

Helper function to create dataset from static data

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| x | npt.NDArray | Required | Numpy data |
| y | npt.NDArray | Required | Numpy labels |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | tf.data.Dataset | tf.data.Dataset: Dataset |

Source: `helia_edge/data/tf_data.py:124`

## helia_edge.data.tf_data.get_output_signature

`function` · `python`

```python
get_output_signature(
    outputs: keras.KerasTensor | npt.NDArray | tuple[keras.KerasTensor | npt.NDArray],
) -> tf.TensorSpec | tuple[tf.TensorSpec, ...]
```

Get output signature from sample outputs

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| outputs | keras.KerasTensor \| npt.NDArray \| tuple[keras.KerasTensor \| npt.NDArray] | Required | Outputs. A tensor or tuple of tensors. Either KerasTensor, tf.Tensor, or numpy array. |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | tf.TensorSpec \| tuple[tf.TensorSpec, ...] | tf.TensorSpec: Tensor spec |

Source: `helia_edge/data/tf_data.py:139`

## helia_edge.data.tf_data.get_output_signature_from_fn

`function` · `python`

```python
get_output_signature_from_fn(
    fn: Callable[..., keras.KerasTensor],
    *args=(),
) -> tf.TensorSpec | tuple[tf.TensorSpec, ...]
```

Get output signature from a function

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| fn | Callable[..., tf.Tensor] | Required | Function |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | tf.TensorSpec \| tuple[tf.TensorSpec, ...] | tf.TensorSpec: Tensor spec |

Source: `helia_edge/data/tf_data.py:165`

## helia_edge.data.tf_data.get_output_signature_from_gen

`function` · `python`

```python
get_output_signature_from_gen(
    gen: Callable[..., Iterator[Any]],
    *args: Any = (),
) -> tf.TensorSpec | tuple[tf.TensorSpec, ...]
```

Get output signature from a generator

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| gen | Callable[..., Iterator[Any]] | Required | Generator factory |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | tf.TensorSpec \| tuple[tf.TensorSpec, ...] | tf.TensorSpec: Tensor spec |

Source: `helia_edge/data/tf_data.py:179`

## helia_edge.data.tf_data.to_tf_dataset

`function` · `python`

```python
to_tf_dataset(dataset: Iterable[Any], output_signature: tf.TensorSpec | tuple | dict[str, Any]) -> tf.data.Dataset
```

Wrap a re-iterable of NumPy batches, such as a Grain dataset, as a ``tf.data.Dataset``.

Elements pass through in order, one ``iter(dataset)`` per pass of the returned dataset;
nothing is batched, shuffled or read in parallel here. Do that in ``dataset``.

**Parameters**

| Name | Type | Default | Description |
| --- | --- | --- | --- |
| dataset | Iterable[Any] | Required | Re-iterable whose ``iter()`` starts from the first element. |
| output_signature | tf.TensorSpec \| tuple \| dict[str, Any] | Required | ``tf.TensorSpec`` structure matching one element. Use ``None`` for a batch dimension that varies, such as a smaller last batch. |

**Returns**

| Name | Type | Description |
| --- | --- | --- |
|  | tf.data.Dataset | tf.data.Dataset: Built with ``tf.data.Dataset.from_generator``. |

**Raises**

| Name | Description |
| --- | --- |
| ImportError | If TensorFlow is not installed (``helia-edge[tensorflow]``). |

Source: `helia_edge/data/tf_data.py:193`