DALI expressions and arithmetic operators

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

[1]:
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.

[2]:
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)

[3]:
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.

[4]:
 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

[5]:
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