<jupyter_start><jupyter_text>ExternalSource operatorIn this example, we will see how to use `ExternalSource` operator which allows us touse an external data source as an input to the Pipeline.<jupyter_code>import types
import collections
import numpy as np
from random import shuffle
from nvidia.dali.pipeline import Pipeline
import nvidia.dali.ops as ops            
import nvidia.dali.types as types

batch_size = 16<jupyter_output><empty_output><jupyter_text>Defining the data sourceIn this example, we use an infinite iterator as a data source.<jupyter_code>class ExternalInputIterator(object):
    def __init__(self, batch_size):
        self.images_dir = "../../data/images/"
        self.batch_size = batch_size
        with open(self.images_dir + "file_list.txt", 'r') as f:
            self.files = [line.rstrip() for line in f if line is not '']
        shuffle(self.files)

    def __iter__(self):
        self.i = 0
        self.n = len(self.files)
        return self

    def __next__(self):
        batch = []
        labels = []
        for _ in range(self.batch_size):
            jpeg_filename, label = self.files[self.i].split(' ')
            f = open(self.images_dir + jpeg_filename, 'rb')
            batch.append(np.frombuffer(f.read(), dtype = np.uint8))
            labels.append(np.array([label], dtype = np.uint8))
            self.i = (self.i + 1) % self.n
        return (batch, labels)<jupyter_output><empty_output><jupyter_text>Defining the pipelineThe next step is to define the Pipeline.The `ExternalSource` operator accepts an iterable or a callable. If the source provides multiple outputs (e.g. images and labels), that number must also be specified as `num_outputs` argument.Internally, the pipeline will call `source` (if callable) or run `next(source)` (if iterable) whenever more data is needed to keep the pipeline running.<jupyter_code>eii = ExternalInputIterator(batch_size)
 class ExternalSourcePipeline(Pipeline):                   
    def __init__(self, batch_size, eii, num_threads, device_id):
        super(ExternalSourcePipeline, self).__init__(batch_size,
                                      num_threads,
                                      device_id,
                                      seed=12)
        
        self.source = ops.ExternalSource(source = eii, num_outputs = 2)
        self.decode = ops.ImageDecoder(device = "mixed", output_type = types.RGB)
        self.enhance = ops.BrightnessContrast(device = "gpu", contrast = 2)

    def define_graph(self):                                                                
        jpegs, labels = self.source()
        images = self.decode(jpegs)
        output = self.enhance(images)
        return (output, labels)<jupyter_output><empty_output><jupyter_text>Using the pipeline<jupyter_code>pipe = ExternalSourcePipeline(batch_size=batch_size, eii=eii, num_threads=2, device_id = 0)
pipe.build()                                                        
pipe_out = pipe.run()<jupyter_output><empty_output><jupyter_text>Notice that labels are still on CPU and no as_cpu call is needed to show them.<jupyter_code>batch_cpu = pipe_out[0].as_cpu()
labels_cpu = pipe_out[1]
from __future__ import print_function
import matplotlib.pyplot as plt
img = batch_cpu.at(2)
print(img.shape)
print(labels_cpu.at(2))
plt.axis('off')
plt.imshow(img)<jupyter_output>(669, 640, 3)
[1]<jupyter_text>Interacting with the GPU inputThe external source operator can also accept GPU data from CuPy or any other data source that supports [cuda array interface](https://numba.pydata.org/numba-doc/latest/cuda/cuda_array_interface.html).For the sake of this example, we will create an ExternalInputGpuIterator in such a way that it returns data on the GPU already. As ImageDecoder doesn't accept data on the GPU we need to decode it outside of DALI on the CPU and then move it to the GPU. In normal cases, the image, or other data would be on the GPU already as a result of the operation of another library.<jupyter_code>import cupy as cp
import imageio

class ExternalInputGpuIterator(object):
    def __init__(self, batch_size):
        self.images_dir = "../../data/images/"
        self.batch_size = batch_size
        with open(self.images_dir + "file_list.txt", 'r') as f:
            self.files = [line.rstrip() for line in f if line is not '']
        shuffle(self.files)

    def __iter__(self):
        self.i = 0
        self.n = len(self.files)
        return self

    def __next__(self):
        batch = []
        labels = []
        for _ in range(self.batch_size):
            jpeg_filename, label = self.files[self.i].split(' ')
            im = imageio.imread(self.images_dir + jpeg_filename)
            im = cp.asarray(im)
            im = im * 0.6;
            batch.append(im.astype(cp.uint8))
            labels.append(cp.array([label], dtype = np.uint8))
            self.i = (self.i + 1) % self.n
        return (batch, labels)<jupyter_output><empty_output><jupyter_text>Let us modify the previous pipeline use the GPU version of ExternalSource operator, removing the decoding as we assume that the raw image is already on the GPU<jupyter_code>eii_gpu = ExternalInputGpuIterator(batch_size)

print(type(next(iter(eii_gpu))[0][0]))
class ExternalSourceGpuPipeline(Pipeline):                   
    def __init__(self, batch_size, eii, num_threads, device_id):
        super(ExternalSourceGpuPipeline, self).__init__(batch_size,
                                      num_threads,
                                      device_id,
                                      seed=12)

        self.source = ops.ExternalSource(device = "gpu", source = eii, num_outputs = 2)
        self.enhance = ops.BrightnessContrast(device = "gpu", contrast = 2)

    def define_graph(self):                                                                
        images, labels = self.source()
        output = self.enhance(images)
        return (output, labels)

pipe_gpu = ExternalSourceGpuPipeline(batch_size=batch_size, eii=eii_gpu, num_threads=2, device_id = 0)
pipe_gpu.build()<jupyter_output><empty_output><jupyter_text>Now let us visualize the results:<jupyter_code>pipe_out_gpu = pipe_gpu.run()
batch_gpu = pipe_out_gpu[0].as_cpu()
labels_gpu = pipe_out_gpu[1].as_cpu()

img = batch_gpu.at(2)
print(img.shape)
print(labels_cpu.at(2))
plt.axis('off')
plt.imshow(img)<jupyter_output>(425, 640, 3)
[1]
{'fraction_non_alphanumeric': 0.09646654158849281, 'fraction_numerical': 0.007817385866166354, 'mean_word_length': 2.7300291545189506, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '373430', 'license': 'ECL-2.0', 'n_tokens_mistral': 1952, 'n_tokens_neox': 1878, 'n_words': 522, 'path': 'docs/examples/general/data_loading/external_input.ipynb', 'repo_name': 'cclauss/DALI', 'size': '6396'}