tf_data

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

Machine-readable model

• convert_inputs_to_tf_datasetfunctionConvert inputs to tf.data.Dataset.
• create_interleaved_dataset_from_generatorfunctionAdapt caller-owned schedules to tf.data without changing sample weights.
• create_dataset_from_datafunctionHelper function to create dataset from static data
• get_output_signaturefunctionGet output signature from sample outputs
• get_output_signature_from_fnfunctionGet output signature from a function
• get_output_signature_from_genfunctionGet output signature from a generator
• to_tf_datasetfunctionWrap a re-iterable of NumPy batches, such as a Grain dataset, as a tf.data.Dataset.

function

convert_inputs_to_tf_dataset

Python

Convert inputs to tf.data.Dataset.

helia_edge/data/tf_data.py:18

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

Convert inputs to tf.data.Dataset.

function

create_interleaved_dataset_from_generator

Python

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

helia_edge/data/tf_data.py:56

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/.

function

create_dataset_from_data

Python

Helper function to create dataset from static data

helia_edge/data/tf_data.py:124

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

Parameters of create_dataset_from_data

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

Returns

Returns of create_dataset_from_data

Type	Description
tf.data.Dataset	tf.data.Dataset: Dataset

function

get_output_signature

Python

Get output signature from sample outputs

helia_edge/data/tf_data.py:139

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

Parameters of get_output_signature

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

Returns of get_output_signature

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

function

get_output_signature_from_fn

Python

Get output signature from a function

helia_edge/data/tf_data.py:165

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

Get output signature from a function

Parameters

Parameters of get_output_signature_from_fn

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

Returns

Returns of get_output_signature_from_fn

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

function

get_output_signature_from_gen

Python

Get output signature from a generator

helia_edge/data/tf_data.py:179

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

Get output signature from a generator

Parameters

Parameters of get_output_signature_from_gen

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

Returns

Returns of get_output_signature_from_gen

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

function

to_tf_dataset

Python

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

helia_edge/data/tf_data.py:193

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

Parameters of to_tf_dataset

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

Returns of to_tf_dataset

Type	Description
tf.data.Dataset	tf.data.Dataset: Built with ``tf.data.Dataset.from_generator``.

Raises

Errors raised by to_tf_dataset

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