Add ExternalSource to dynamic mode

[rostan-t]

Category:

New feature (non-breaking change which adds functionality)

Description:

Add ExternalSource to dynamic mode. It mimics fn.external_source to some extent but behaves differently, especially with batches, to integrate well with the imperative API.

This is not a very useful operator in and of itself but will enable sources other than a reader in transparent pipelining.

Additional information:

Affected modules and functionalities:

Dynamic mode.

Key points relevant for the review:

Tests:

• Existing tests apply
• New tests added
  • Python tests
  • GTests
  • Benchmark
  • Other
• N/A

Checklist

Documentation

• Existing documentation applies
• Documentation updated
  • Docstring
  • Doxygen
  • RST
  • Jupyter
  • Other
• N/A

ndd.ExternalSource doesn't appear in the documentation yet because it's only really useful with transparent pipelining.

DALI team only

Requirements

• Implements new requirements
• Affects existing requirements
• N/A

REQ IDs: N/A

JIRA TASK: DALI-4745

[github-advanced-security]

CodeQL found more than 20 potential problems in the proposed changes. Check the Files changed tab for more details.

[greptile-apps]

Greptile Summary

This PR adds ndd.ExternalSource to the dynamic (imperative) API, allowing data to be fed into dynamic pipelines from Python callables or iterables. It mirrors fn.external_source but integrates with the imperative batch model — the caller invokes the object directly and can pass batch_size to broadcast a single-sample output into a full batch.

• _external_source.py introduces the ExternalSource class, delegating source handling to the existing get_callback_from_source utility and output conversion to as_tensor/as_batch/Batch.broadcast. The _broadcast_arg helper correctly treats strings as scalars, and _are_types_uniform guards against mixed Tensor/Batch outputs across multi-output sources.
• test_external_source.py covers callable, iterable, generator, batch-output, broadcast, cycle modes (no/quiet/raise), layout/dtype propagation, and CPU/GPU device placement; the assert_raises globs use the case-insensitive default, so they correctly match the capitalised error messages.
• One-line __init__.py change exports ExternalSource into the ndd namespace.

Confidence Score: 5/5

Safe to merge; the new ExternalSource class delegates to well-tested existing utilities and introduces no observable regressions in the dynamic API.

The implementation correctly handles all documented source kinds (callable, iterable, generator function), cycle modes, and multi-output routing. The _broadcast_arg string-sentinel, _get_batch_size dispatch, and _are_types_uniform uniformity check are all logically sound. The test suite covers the main paths, edge cases, and error conditions. The one observation left is a minor error-message improvement with no correctness impact.

No files require special attention.

Important Files Changed

Filename	Overview
dali/python/nvidia/dali/experimental/dynamic/_external_source.py	New ExternalSource class for dynamic mode — wraps get_callback_from_source, dispatches to as_tensor/as_batch, and handles broadcasting. Logic is sound; one minor error message nit.
dali/test/python/experimental_mode/test_external_source.py	Comprehensive tests: callable, iterable, generator, batch output, broadcast, cycle modes, layout/dtype propagation, and device placement. Edge cases well covered.
dali/python/nvidia/dali/experimental/dynamic/init.py	One-line addition exporting ExternalSource into the ndd namespace.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["ExternalSource.__call__(batch_size)"] --> B["_callback()"]
    B --> C["_get_outputs(data)"]
    C -->|"num_outputs == 1"| D["wrap data in 1-tuple"]
    C -->|"num_outputs > 1"| E{"isinstance(data, Sequence)\nand len == num_outputs?"}
    E -->|"No"| F["raise ValueError"]
    E -->|"Yes"| G["return data as-is"]
    D --> H["_convert_output(item, batch_size, idx)"]
    G --> H
    H --> I{"_get_batch_size(data)?"}
    I -->|"not None (Batch/TensorList)"| J["as_batch(data)"]
    I -->|"None (tensor-like)"| K["as_tensor(data)"]
    K -->|"batch_size provided"| L["Batch.broadcast(tensor, batch_size)"]
    K -->|"no batch_size"| M["return Tensor"]
    J --> N["validate batch_size"]
    N --> O["return Batch"]
    L --> O
    M --> P["_are_types_uniform(results)"]
    O --> P
    P -->|"mixed types"| Q["raise TypeError"]
    P -->|"uniform"| R["return results[0] or tuple"]

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
    A["ExternalSource.__call__(batch_size)"] --> B["_callback()"]
    B --> C["_get_outputs(data)"]
    C -->|"num_outputs == 1"| D["wrap data in 1-tuple"]
    C -->|"num_outputs > 1"| E{"isinstance(data, Sequence)\nand len == num_outputs?"}
    E -->|"No"| F["raise ValueError"]
    E -->|"Yes"| G["return data as-is"]
    D --> H["_convert_output(item, batch_size, idx)"]
    G --> H
    H --> I{"_get_batch_size(data)?"}
    I -->|"not None (Batch/TensorList)"| J["as_batch(data)"]
    I -->|"None (tensor-like)"| K["as_tensor(data)"]
    K -->|"batch_size provided"| L["Batch.broadcast(tensor, batch_size)"]
    K -->|"no batch_size"| M["return Tensor"]
    J --> N["validate batch_size"]
    N --> O["return Batch"]
    L --> O
    M --> P["_are_types_uniform(results)"]
    O --> P
    P -->|"mixed types"| Q["raise TypeError"]
    P -->|"uniform"| R["return results[0] or tuple"]

Loading

_{Reviews (4): Last reviewed commit: "Add tests for ndd.ExternalSource" | Re-trigger Greptile}

[rostan-t]

!build

[dali-automaton]

CI MESSAGE: [55431712]: BUILD STARTED

[mzient]

Not particularly descriptive.

[rostan-t]

Is this better now?

Consume data from a Python callable or iterable source.

[dali-automaton]

CI MESSAGE: [55431712]: BUILD FAILED

[mzient]

Shouldn't ExternalSource have next_epoch to be usable like readers? Somewhat tangentially - is raise/end of epoch properly handle by pipelines?

[rostan-t]

In normal dynamic mode, ExternalSource is not particularly useful so I deliberately enabled use only through the __call__ method.

When integrating with transparent pipelining, I'm adding a method .compiled(batch_size) that's usable specifically for this. The reason it's not next_epoch is that the notion of an epoch doesn't necessarily make sense for ExternalSource. This depend on the nature of the source and the cycle argument.

---

Somewhat tangentially - is raise/end of epoch properly handle by pipelines?

In pipeline mode, this really depends on cycle. To create epochs cycle="raise" is used.

[rostan-t]

!build

[dali-automaton]

CI MESSAGE: [55574360]: BUILD STARTED

[dali-automaton]

CI MESSAGE: [55574360]: BUILD PASSED