DALI expressions and arithmetic operators

In this example, we will see how to use arithmetic operators in DALI Pipeline.

import types
import collections
import numpy as np
from nvidia.dali.pipeline import Pipeline
import nvidia.dali.ops as ops
import nvidia.dali.types as types

batch_size = 1

Defining the iterator

We will use custom iterator producing small tensors filled with series of numbers, so we can easily inspect the results.

class ExternalInputIterator(object):
    def __init__(self, batch_size, left_type, right_type):
        self.batch_size = batch_size
        self.left_type = left_type
        self.right_type = right_type

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

    def __next__(self):
        left = []
        right = []
        for sample in range(self.batch_size):
            left.append(np.array([sample + self.i], dtype = self.left_type))
            right.append(np.array([self.batch_size + self.i + sample], dtype = self.right_type))
        self.i = (self.i + 1) % self.n
        return (left, right)

    next = __next__

Instantiating the iterators

We create instances of the ExternalInputIterator with different type combinations. Type promotions for binary operators are described below. They apply to +, -, * and //. The / always returns a float32 for integer inputs, and applies the rules below when at least one of the inputs is a floating point number.

T      op T      = T   floatX op T      = floatX           (where T is not a float)   floatX op floatY = float(max(X, Y))   intX   op intY   = int(max(X, Y))   uintX  op uintY  = uint(max(X, Y))   intX   op uintY  = int2Y            (if X <= Y)   intX   op uintY  = intX             (if X > Y)

iterator_u8_u8 = iter(ExternalInputIterator(batch_size, np.uint8, np.uint8))
iterator_u8_i32 = iter(ExternalInputIterator(batch_size, np.uint8, np.int32))
iterator_i16_u8 = iter(ExternalInputIterator(batch_size, np.int16, np.uint8))
iterator_i32_f32 = iter(ExternalInputIterator(batch_size, np.int32, np.float32))

Defining the pipeline

The next step is to define the Pipeline.

We override Pipeline.iter_setup, a method called by the pipeline before every Pipeline.run, to call the iterator and feed the result to ExternalSource() operators, referenced by self.left and self.right, by using feed_input.

Note, that we do not need to instantiate any additional operators, we can use regular Python arithmetic expression on the results of other operators in the define_graph step.

Here we return both of the inputs and the result of self.right + self.right * self.left.

 class ExternalSourcePipeline(Pipeline):
    def __init__(self, iterator, batch_size, num_threads, device_id):
        super(ExternalSourcePipeline, self).__init__(batch_size, num_threads, device_id, seed=12)
        self.left_source = ops.ExternalSource()
        self.right_source = ops.ExternalSource()
        self.iterator = iterator

    def define_graph(self):
        self.left = self.left_source()
        self.right = self.right_source()
        return self.left, self.right, self.right + self.right * self.left

    def iter_setup(self):
        (l, r) = self.iterator.next()
        self.feed_input(self.left, l)
        self.feed_input(self.right, r)

Using the pipeline

for it in [iterator_u8_u8, iterator_u8_i32, iterator_i16_u8, iterator_i32_f32]:
    pipe = ExternalSourcePipeline(it, batch_size=batch_size, num_threads=2, device_id = 0)
    pipe.build()
    pipe_out = pipe.run()
    l = pipe_out[0].as_array()
    r = pipe_out[1].as_array()
    out = pipe_out[2].as_array()
    print("{} + {} * {} = {} of type {}".format(r, r, l, out, out.dtype))

[[1]] + [[1]] * [[0]] = [[1]] of type uint8
[[1]] + [[1]] * [[0]] = [[1]] of type int32
[[1]] + [[1]] * [[0]] = [[1]] of type int16
[[1.]] + [[1.]] * [[0]] = [[1.]] of type float32