()
| 64 | |
| 65 | |
| 66 | def test_argument_input(): |
| 67 | # execution environment |
| 68 | tp = _b._ThreadPool(4) |
| 69 | stream = _b.Stream(0) |
| 70 | |
| 71 | # standalone operator invocation |
| 72 | # mock inputs |
| 73 | a = dali.data_node.DataNode("a", "cpu", None) |
| 74 | b = dali.data_node.DataNode("b", "cpu", None) |
| 75 | c = dali.data_node.DataNode("c", "cpu", None) |
| 76 | # the operation - in this case an arithmetic operation |
| 77 | x = fn.normalize(a, mean=b, stddev=c) |
| 78 | # the spec of the operator |
| 79 | spec = x.source.spec |
| 80 | # complete the spec with the execution environment |
| 81 | complete_spec(spec, max_batch_size=10) # oversized max batch size |
| 82 | # create the operator |
| 83 | op = _b._Operator(spec) |
| 84 | # create the workspace and populat ethe environment |
| 85 | ws = _b._Workspace(tp) |
| 86 | ws.SetStream(stream) |
| 87 | # actual inputs |
| 88 | A = dali.tensors.TensorListCPU( |
| 89 | [ |
| 90 | np.int32([4, 5, 6]), |
| 91 | np.int32([6, 2, -2]), |
| 92 | np.int32([13, 18, 23]), |
| 93 | ] |
| 94 | ) |
| 95 | B = dali.tensors.TensorListCPU([np.float32([1]), np.float32([2]), np.float32([3])]) |
| 96 | C = dali.tensors.TensorListCPU([np.float32([1]), np.float32([4]), np.float32([5])]) |
| 97 | ws.AddInput(A) |
| 98 | ws.AddArgumentInput("mean", B) |
| 99 | ws.AddArgumentInput("stddev", C) |
| 100 | # run the operator |
| 101 | op.SetupAndRun(ws, batch_size=3) |
| 102 | # get the output |
| 103 | (out,) = ws.GetOutputs() |
| 104 | np.array_equal(out[0], np.float32([3, 4, 5])) |
| 105 | np.array_equal(out[1], np.float32([1, 0, -1])) |
| 106 | np.array_equal(out[2], np.float32([2, 3, 4])) |
| 107 | |
| 108 | |
| 109 | dali_extra_path = os.environ.get("DALI_EXTRA_PATH", None) |
nothing calls this directly
no test coverage detected