Basic implementation of compute_test_value mechanism

theano/configdefaults.py

@@ -200,3 +200,9 @@ AddConfigVar('warn.sum_sum_bug',
 AddConfigVar('warn.sum_div_dimshuffle_bug',
              "Warn if previous versions of Theano (between rev. 3bd9b789f5e8, 2010-06-16, and cfc6322e5ad4, 2010-08-03) would have given incorrect result. This bug was triggered by sum of division of dimshuffled tensors.",
              BoolParam(default_0_3))
+
+AddConfigVar('compute_test_value',
+        "If True, Theano will run each op at graph build time, using Constants, SharedVariables and the tag 'test_value' as inputs to the function. This helps the user track down problems in the graph before it gets optimized.",
+        EnumStr(True, False, 'warn', 'err'))
+
+

theano/gof/op.py

@@ -8,9 +8,10 @@ compatible with `gof`'s :doc:`graph` routines.
 
 __docformat__ = "restructuredtext en"
 
+from .. import config
+import graph
+import numpy
 import utils
-from theano import config
-
 
 
 class CLinkerObject(object):
@@ -322,6 +323,46 @@ class PureOp(object):
         """
         node = self.make_node(*inputs, **kwargs)
         self.add_tag_trace(node)
+    
+        if config.compute_test_value:
+            # avoid circular import
+            from ..compile.sharedvalue import SharedVariable
+            run_perform = True
+
+            # build test input-values
+            input_vals = []
+            for ins in inputs:
+                if isinstance(ins, graph.Constant):
+                    input_vals.append(ins.value)
+                elif isinstance(ins,numpy.ndarray):
+                    input_vals.append(ins)
+                elif isinstance(ins,SharedVariable):
+                    input_vals.append(ins.get_value(borrow=True))
+                elif isinstance(ins,graph.Variable) and hasattr(ins.tag, 'test_value'):
+                    input_vals.append(ins.tag.test_value)
+                else:
+                    # no test-value was specified, act accordingly
+                    if config.compute_test_value == 'warn':
+                        raise Warning('Cannot compute test value: input %s of Op %s missing default value')
+                        run_perform = False
+                    elif config.compute_test_value == 'err':
+                        raise ValueError('Cannot compute test value: input %s of Op %s missing default value')
+                    else:
+                        # silently skip test
+                        run_perform = False
+          
+            # if all inputs have test-values, run the actual op
+            if run_perform:
+
+                # compute output value once with test inputs to validate graph
+                output_storage = [[None] * len(node.outputs)]
+                node.op.perform(node, input_vals, output_storage)
+           
+                # add 'test_value' to output tags, so that downstream ops can use these
+                # numerical values as inputs to their perform method.
+                for (outval, node_output) in zip(output_storage, node.outputs):
+                    node_output.tag.test_value = outval[0]
+
         if self.default_output is not None:
             return node.outputs[self.default_output]
         else:

theano/gof/tests/test_compute_test_value.py

+import numpy
+import unittest
+
+import theano
+from theano import tensor as T
+
+
+class TestComputeTestValue(unittest.TestCase):
+
+    def test_variable_only(self):
+        theano.config.compute_test_value = True
+
+        x = T.matrix('x')
+        x.tag.test_value = numpy.random.rand(3,4)
+        y = T.matrix('y')
+        y.tag.test_value = numpy.random.rand(4,5)
+        
+        # should work
+        z = T.dot(x,y)
+
+        # this test should fail
+        y.tag.test_value = numpy.random.rand(6,5)
+        self.assertRaises(ValueError, T.dot, x, y)
+
+    def test_compute_flag(self):
+
+        x = T.matrix('x')
+        y = T.matrix('y')
+        y.tag.test_value = numpy.random.rand(4,5)
+        
+        # should skip computation of test value
+        theano.config.compute_test_value = False
+        z = T.dot(x,y)
+
+        # should fail one or another when flag is set
+        theano.config.compute_test_value = 'warn'
+        self.assertRaises(Warning, T.dot, x, y)
+        theano.config.compute_test_value = 'err'
+        self.assertRaises(ValueError, T.dot, x, y)
+
+    def test_string_var(self):
+        theano.config.compute_test_value = True
+
+        x = T.matrix('x')
+        x.tag.test_value = numpy.random.rand(3,4)
+        y = T.matrix('y')
+        y.tag.test_value = numpy.random.rand(4,5)
+        
+        z = theano.shared(numpy.random.rand(5,6))
+
+        # should work
+        out = T.dot(T.dot(x,y), z)
+        def f(x,y,z):
+            return T.dot(T.dot(x,y),z)
+
+        # this test should fail
+        z.set_value(numpy.random.rand(7,6))
+        self.assertRaises(ValueError, f, x, y, z)
+
+    def test_shared(self):
+        theano.config.compute_test_value = True
+
+        x = T.matrix('x')
+        x.tag.test_value = numpy.random.rand(3,4)
+        y = theano.shared(numpy.random.rand(4,6), 'y')
+        
+        # should work
+        z = T.dot(x,y)
+
+        # this test should fail
+        y.set_value(numpy.random.rand(5,6))
+        self.assertRaises(ValueError, T.dot, x, y)
+
+    def test_ndarray(self):
+        theano.config.compute_test_value = True
+
+        x = numpy.random.rand(2,3)
+        y = theano.shared(numpy.random.rand(3,6), 'y')
+        
+        # should work
+        z = T.dot(x,y)
+
+        # this test should fail
+        x = numpy.random.rand(2,4)
+        self.assertRaises(ValueError, T.dot, x, y)
+
+    def test_constant(self):
+        theano.config.compute_test_value = True
+
+        x = T.constant(numpy.random.rand(2,3))
+        y = theano.shared(numpy.random.rand(3,6), 'y')
+        
+        # should work
+        z = T.dot(x,y)
+
+        # this test should fail
+        x = T.constant(numpy.random.rand(2,4))
+        self.assertRaises(ValueError, T.dot, x, y)