/usr/local/lib64/python3.6/site-packages/caffe2/python
Edit: /usr/local/lib64/python3.6/site-packages/caffe2/python/python_op_test.py (9169B)
from caffe2.python import core, workspace
from caffe2.python.core import CreatePythonOperator
import caffe2.python.hypothesis_test_util as hu
from hypothesis import given, settings
import hypothesis.strategies as st
import numpy as np
class CustomError(Exception):
pass
def SubFunctionThatThrowsCustomError():
raise CustomError("This is an intentional exception.")
def MainOpFunctionThatThrowsCustomError(inputs, _):
return SubFunctionThatThrowsCustomError()
def MainOpFunctionThatThrowsCustomErrorInBuilder(inputs, _):
raise CustomError("This is an intentional exception in builder.")
def op_builder(name, index, extra):
iterations = [0]
assert name == 'name'
assert index == 5
assert extra - 4.2 < 0.0001
def my_op(inputs, outputs):
assert inputs[0].data[0] == iterations[0]
assert name == 'name'
assert index == 5
assert extra - 4.2 < 0.0001
iterations[0] += 1
return my_op
class PythonOpTest(hu.HypothesisTestCase):
@given(x=hu.tensor())
def test_feed(self, x):
def f(inputs, _):
self.assertEqual(x.shape, inputs[0].shape)
self.assertEqual(type(inputs[0].shape), tuple)
self.assertEqual(type(inputs[0].data), np.ndarray)
np.testing.assert_almost_equal(x, inputs[0].data)
op = CreatePythonOperator(f, ["x"], [])
workspace.FeedBlob("x", x)
workspace.RunOperatorOnce(op)
def test_exception(self):
op = CreatePythonOperator(MainOpFunctionThatThrowsCustomError, [], [])
with self.assertRaisesRegex(CustomError, "This is an intentional exception."):
workspace.RunOperatorOnce(op)
def test_exception_builder(self):
op = CreatePythonOperator(MainOpFunctionThatThrowsCustomErrorInBuilder, [], [])
with self.assertRaisesRegex(CustomError, "This is an intentional exception in builder."):
workspace.RunOperatorOnce(op)
@given(x=hu.tensor())
def test_feed_with_helper_function(self, x):
def f(inputs, _):
self.assertEqual(x.shape, inputs[0].shape)
self.assertEqual(type(inputs[0].shape), tuple)
self.assertEqual(type(inputs[0].data), np.ndarray)
np.testing.assert_almost_equal(x, inputs[0].data)
net = core.Net("test")
net.Python(f)(["x"], [])
workspace.FeedBlob("x", x)
workspace.RunNetOnce(net)
def test_builder_tuple(self):
net = core.Net("builder_template")
iter_blob = 'iter'
net.Python((op_builder, ['name', 5], {'extra': 4.2}))([iter_blob], [])
net.Python((op_builder, ['name', 5], {'extra': 4.2}))([iter_blob], [])
for repeat in range(2):
# check that the builder will be called exactly once for each
# PythonOp constructor. Cloning the net will also trigger a call
# to the builder when the net is created.
cloned_net = net.Clone('builder_%d' % repeat)
workspace.FeedBlob(iter_blob, np.array([0]))
# Builder gets called once per python op in the line below
workspace.CreateNet(cloned_net)
for i in range(10):
workspace.FeedBlob(iter_blob, np.array([i]))
workspace.RunNet(cloned_net)
@given(x=hu.tensor())
def test_feed_with_gc(self, x):
def f(inputs, _):
self.assertEqual(x.shape, inputs[0].shape)
np.testing.assert_almost_equal(x, inputs[0].data)
op = CreatePythonOperator(f, ["x"], [])
workspace.FeedBlob("x", x)
workspace.RunOperatorOnce(op)
del f
workspace.FeedBlob("x", x)
workspace.RunOperatorOnce(op)
@given(x=hu.tensor())
def test_reshape(self, x):
def f(inputs, outputs):
outputs[0].reshape(inputs[0].shape)
self.assertEqual(x.shape, inputs[0].shape)
self.assertEqual(x.shape, outputs[0].shape)
outputs[0].data[...] = inputs[0].data
op = CreatePythonOperator(f, ["x"], ["y"])
workspace.FeedBlob("x", x)
workspace.RunOperatorOnce(op)
y = workspace.FetchBlob("y")
np.testing.assert_almost_equal(x, y)
@given(x=hu.tensor())
def test_workspace_manipulation(self, x):
"""
Verify that python op can manipulate workspace directly
"""
def f(inputs, outputs, ws):
fetched = ws.blobs['internal'].fetch()
np.testing.assert_almost_equal(fetched, x)
ws = workspace.C.Workspace()
net = core.Net("test")
net.GivenTensorFill([], ['internal'], values=x, shape=x.shape)
net.Python(f, pass_workspace=True)([], [])
ws.run(net)
@given(x=hu.tensor())
def test_caught_exception_doesnt_terminate(self, x):
def f(inputs, outputs):
try:
raise Exception("Exception in handler")
except Exception:
pass
op = CreatePythonOperator(f, ["x"], ["y"])
workspace.FeedBlob("x", x)
workspace.RunOperatorOnce(op)
@given(x=hu.tensor(),
n=st.integers(min_value=1, max_value=20),
w=st.integers(min_value=1, max_value=20))
@settings(deadline=1000)
def test_multithreaded_evaluation(self, x, n, w):
def f(inputs, outputs):
outputs[0].reshape(inputs[0].shape)
outputs[0].data[...] = inputs[0].data
ops = [CreatePythonOperator(f, ["x"], [str(i)]) for i in range(n)]
net = core.Net("net")
net.Proto().op.extend(ops)
net.Proto().type = "dag"
net.Proto().num_workers = w
iters = 100
plan = core.Plan("plan")
plan.AddStep(core.ExecutionStep("test-step", net, iters))
workspace.FeedBlob("x", x)
workspace.RunPlan(plan.Proto().SerializeToString())
for i in range(n):
y = workspace.FetchBlob(str(i))
np.testing.assert_almost_equal(x, y)
@given(x=hu.tensor(), in_place=st.booleans(), **hu.gcs)
@settings(deadline=10000)
def test_gradient(self, x, in_place, gc, dc):
def f(inputs, outputs):
outputs[0].reshape(inputs[0].shape)
outputs[0].data[...] = inputs[0].data * 2
def grad_f(inputs, outputs):
# Ordering is [inputs, outputs, grad_outputs]
grad_output = inputs[2]
grad_input = outputs[0]
grad_input.reshape(grad_output.shape)
grad_input.data[...] = grad_output.data * 2
op = CreatePythonOperator(
f, ["x"], ["x" if in_place else "y"], grad_f=grad_f)
self.assertGradientChecks(gc, op, [x], 0, [0])
self.assertDeviceChecks(dc, op, [x], [0])
@given(inputs=hu.tensors(n=2), **hu.gcs)
@settings(deadline=10000)
def test_gradient_multiple(self, inputs, gc, dc):
(x1, x2) = inputs
def f(inputs, outputs):
for idx in [0, 1]:
self.assertEqual(type(inputs[idx].shape), tuple)
outputs[idx].reshape(inputs[idx].shape)
outputs[idx].data[...] = inputs[idx].data * 2
def grad_f(inputs, outputs):
# Ordering is [inputs, outputs, grad_outputs]
self.assertEqual(len(inputs), 6)
self.assertEqual(len(outputs), 2)
for (grad_output_idx, grad_input_idx) in [(4, 0), (5, 1)]:
grad_output = inputs[grad_output_idx]
grad_input = outputs[grad_input_idx]
grad_input.reshape(grad_output.shape)
grad_input.data[...] = grad_output.data * 2
op = CreatePythonOperator(f, ["x1", "x2"], ["y1", "y2"], grad_f=grad_f)
for idx in [0, 1]:
self.assertGradientChecks(gc, op, [x1, x2], idx, [0, 1])
self.assertDeviceChecks(dc, op, [x1, x2], [0, 1])
@given(inputs=hu.tensors(n=3), **hu.gcs)
@settings(deadline=10000)
def test_gradient_multiple_with_indices(self, inputs, gc, dc):
(x1, x2, x3) = inputs
def f(inputs, outputs):
for idx in [0, 1, 2]:
self.assertEqual(type(inputs[idx].shape), tuple)
outputs[idx].reshape(inputs[idx].shape)
outputs[idx].data[...] = inputs[idx].data * 2
def grad_f(inputs, outputs):
# Ordering is [inputs, outputs, grad_outputs]
self.assertEqual(len(inputs), 8)
self.assertEqual(len(outputs), 1)
for (grad_output_idx, grad_input_idx) in [(6, 0)]:
grad_output = inputs[grad_output_idx]
grad_input = outputs[grad_input_idx]
grad_input.reshape(grad_output.shape)
grad_input.data[...] = grad_output.data * 2
op = CreatePythonOperator(
f, ["x1", "x2", "x3"], ["y1", "y2", "y3"],
grad_f=grad_f,
grad_output_indices=[0, 2], # Receive grad outputs for y1 and y3
grad_input_indices=[0] # Produce grad inputs for x1
)
self.assertGradientChecks(gc, op, [x1, x2, x3], 0, [0, 2])
self.assertDeviceChecks(dc, op, [x1, x2, x3], [0, 1, 2])