Merge pull request #1862 from Hengjean/TypedList

theano/tensor/type_other.py

@@ -2,7 +2,7 @@
 # Slice type and Op. None Type and NoneConst.
 #
 import theano
-from theano.gof import Apply, Constant, Generic, Op, Type
+from theano.gof import Apply, Constant, Generic, Op, Type, hashtype
 from theano.gradient import DisconnectedType
 
 
@@ -52,6 +52,12 @@ class SliceType(Type):
     def __str__(self):
         return "slice"
 
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hashtype(self)
+
 slicetype = SliceType()
 
 

theano/typed_list/__init__.py

+from type import TypedListType
+from basic import *

theano/typed_list/basic.py

+from type import TypedListType
+
+from theano.gof import Apply, Constant, Op, Variable
+from theano.tensor.type_other import SliceType
+from theano import tensor as T
+
+import numpy
+
+
+class _typed_list_py_operators:
+
+    def __getitem__(self, index):
+        return GetItem()(self, index)
+
+    def append(self, toAppend):
+        return Append()(self, toAppend)
+
+    def extend(self, toAppend):
+        return Extend()(self, toAppend)
+
+    ttype = property(lambda self: self.type.ttype)
+
+
+class TypedListVariable(_typed_list_py_operators, Variable):
+    """
+    Subclass to add the typed list operators to the basic `Variable` class.
+    """
+
+TypedListType.Variable = TypedListVariable
+
+
+class GetItem(Op):
+    """
+    get specified slice of a typed list
+    """
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hash(type(self))
+
+    def make_node(self, x, index):
+        assert isinstance(x.type, TypedListType)
+        if not isinstance(index, Variable):
+            if isinstance(index, slice):
+                index = Constant(SliceType(), index)
+                return Apply(self, [x, index], [x.type()])
+            else:
+                index = T.constant(index, ndim=0)
+                return Apply(self, [x, index], [x.ttype()])
+        if isinstance(index.type, SliceType):
+            return Apply(self, [x, index], [x.type()])
+        elif isinstance(index, T.TensorVariable) and index.ndim == 0:
+            return Apply(self, [x, index], [x.ttype()])
+        else:
+            raise TypeError('Expected scalar or slice as index.')
+
+    def perform(self, node, (x, index), (out, )):
+        if not isinstance(index, slice):
+            index = int(index)
+        out[0] = x[index]
+
+    def __str__(self):
+        return self.__class__.__name__
+
+
+class Append(Op):
+    """
+    #append an element at the end of another list
+    """
+
+    def __init__(self, inplace=False):
+        self.inplace = inplace
+        if self.inplace:
+            self.destroy_map = {0: [0]}
+
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hash(type(self))
+
+    def make_node(self, x, toAppend):
+        assert isinstance(x.type, TypedListType)
+        assert x.ttype == toAppend.type
+        return Apply(self, [x, toAppend], [x.type()])
+
+    def perform(self, node, (x, toAppend), (out, )):
+        if not self.inplace:
+            out[0] = list(x)
+        else:
+            out[0] = x
+        out[0].append(toAppend)
+
+    def __str__(self):
+        return self.__class__.__name__
+
+
+class Extend(Op):
+    """
+    append all element of a list at the end of another list
+    """
+
+    def __init__(self, inplace=False):
+        self.inplace = inplace
+        if self.inplace:
+            self.destroy_map = {0: [0]}
+
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hash(type(self))
+
+    def make_node(self, x, toAppend):
+        assert isinstance(x.type, TypedListType)
+        assert x.type == toAppend.type
+        return Apply(self, [x, toAppend], [x.type()])
+
+    def perform(self, node, (x, toAppend), (out, )):
+        if not self.inplace:
+            out[0] = list(x)
+        else:
+            out[0] = x
+        out[0].extend(toAppend)
+
+    def __str__(self):
+        return self.__class__.__name__
+
+
+class Insert(Op):
+
+    def __init__(self, inplace=False):
+        self.inplace = inplace
+        if self.inplace:
+            self.destroy_map = {0: [0]}
+
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hash(type(self))
+
+    def make_node(self, x, index, toInsert):
+        assert isinstance(x.type, TypedListType)
+        assert x.ttype == toInsert.type
+        if not isinstance(index, Variable):
+            index = T.constant(index, ndim=0)
+        else:
+            assert isinstance(index, T.TensorVariable) and index.ndim == 0
+        return Apply(self, [x, index, toInsert], [x.type()])
+
+    def perform(self, node, (x, index, toInsert), (out, )):
+        if not self.inplace:
+            out[0] = list(x)
+        else:
+            out[0] = x
+        out[0].insert(index, toInsert)
+
+    def __str__(self):
+        return self.__class__.__name__
+
+
+class Remove(Op):
+
+    def __init__(self, inplace=False):
+        self.inplace = inplace
+        if self.inplace:
+            self.destroy_map = {0: [0]}
+
+    def __eq__(self, other):
+        return type(self) == type(other)
+
+    def __hash__(self):
+        return hash(type(self))
+
+    def make_node(self, x, toRemove):
+        assert isinstance(x.type, TypedListType)
+        assert x.ttype == toRemove.type
+        return Apply(self, [x, toRemove], [x.type()])
+
+    def perform(self, node, (x, toRemove), (out, )):
+
+        if not self.inplace:
+            out[0] = list(x)
+        else:
+            out[0] = x
+
+        """
+        inelegant workaround for ValueError: The truth value of an
+        array with more than one element is ambiguous. Use a.any() or a.all()
+        being thrown when trying to remove a matrix from a matrices list
+        """
+        if isinstance(toRemove, numpy.ndarray):
+            for y in range(out[0].__len__()):
+                if numpy.array_equal(out[0][y], toRemove):
+                    del out[0][y]
+                    break
+        else:
+            out[0].remove(toRemove)
+
+    def __str__(self):
+        return self.__class__.__name__

theano/typed_list/tests/test_basic.py

+import unittest
+
+import numpy
+
+import theano
+import theano.typed_list
+from theano import tensor as T
+from theano.tensor.type_other import SliceType
+from theano.typed_list.type import TypedListType
+from theano.typed_list.basic import (GetItem, Insert,
+                                      Append, Extend, Remove)
+from theano.tests import unittest_tools as utt
+
+
+#took from tensors/tests/test_basic.py
+def rand_ranged_matrix(minimum, maximum, shape):
+    return numpy.asarray(numpy.random.rand(*shape) * (maximum - minimum)
+                         + minimum, dtype=theano.config.floatX)
+
+
+class test_get_item(unittest.TestCase):
+
+    def setUp(self):
+        utt.seed_rng()
+
+    def test_sanity_check_slice(self):
+
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        mySymbolicSlice = SliceType()()
+
+        z = GetItem()(mySymbolicMatricesList, mySymbolicSlice)
+
+        self.assertFalse(isinstance(z, T.TensorVariable))
+
+        f = theano.function([mySymbolicMatricesList, mySymbolicSlice],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], slice(0, 1, 1)), [x]))
+
+    def test_sanity_check_single(self):
+
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        mySymbolicScalar = T.scalar()
+
+        z = GetItem()(mySymbolicMatricesList, mySymbolicScalar)
+
+        f = theano.function([mySymbolicMatricesList, mySymbolicScalar],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], numpy.asarray(0)), x))
+
+    def test_interface(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        mySymbolicScalar = T.scalar()
+
+        z = mySymbolicMatricesList[mySymbolicScalar]
+
+        f = theano.function([mySymbolicMatricesList, mySymbolicScalar],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], numpy.asarray(0)), x))
+
+        z = mySymbolicMatricesList[0: 1: 1]
+
+        f = theano.function([mySymbolicMatricesList],
+                            z)
+
+        self.assertTrue(numpy.array_equal(f([x]), [x]))
+
+    def test_wrong_input(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        mySymbolicMatrix = T.matrix()
+
+        self.assertRaises(TypeError, GetItem(), mySymbolicMatricesList,
+                          mySymbolicMatrix)
+
+    def test_constant_input(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        z = GetItem()(mySymbolicMatricesList, 0)
+
+        f = theano.function([mySymbolicMatricesList],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x]), x))
+
+        z = GetItem()(mySymbolicMatricesList, slice(0, 1, 1))
+
+        f = theano.function([mySymbolicMatricesList],
+                            z)
+
+        self.assertTrue(numpy.array_equal(f([x]), [x]))
+
+
+class test_append(unittest.TestCase):
+
+    def test_inplace(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+
+        z = Append(True)(mySymbolicMatricesList, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myMatrix], z,
+                            accept_inplace=True)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], y), [x, y]))
+
+    def test_sanity_check(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                                theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+
+        z = Append()(mySymbolicMatricesList, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myMatrix], z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], y), [x, y]))
+
+    def test_interfaces(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                                theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+
+        z = mySymbolicMatricesList.append(myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myMatrix], z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], y), [x, y]))
+
+
+class test_extend(unittest.TestCase):
+
+    def test_inplace(self):
+        mySymbolicMatricesList1 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        mySymbolicMatricesList2 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        z = Extend(True)(mySymbolicMatricesList1, mySymbolicMatricesList2)
+
+        f = theano.function([mySymbolicMatricesList1, mySymbolicMatricesList2],
+                            z, accept_inplace=True)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], [y]), [x, y]))
+
+    def test_sanity_check(self):
+        mySymbolicMatricesList1 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        mySymbolicMatricesList2 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        z = Extend()(mySymbolicMatricesList1, mySymbolicMatricesList2)
+
+        f = theano.function([mySymbolicMatricesList1, mySymbolicMatricesList2],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], [y]), [x, y]))
+
+    def test_interface(self):
+        mySymbolicMatricesList1 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        mySymbolicMatricesList2 = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+
+        z = mySymbolicMatricesList1.extend(mySymbolicMatricesList2)
+
+        f = theano.function([mySymbolicMatricesList1, mySymbolicMatricesList2],
+                            z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], [y]), [x, y]))
+
+
+class test_insert(unittest.TestCase):
+
+    def test_inplace(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+        myScalar = T.scalar()
+
+        z = Insert(True)(mySymbolicMatricesList, myScalar, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myScalar, myMatrix], z,
+                            accept_inplace=True)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], numpy.asarray(1), y), [x, y]))
+
+    def test_sanity_check(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                                theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+        myScalar = T.scalar()
+
+        z = Insert()(mySymbolicMatricesList, myScalar, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myScalar, myMatrix], z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x], numpy.asarray(1), y), [x, y]))
+
+
+class test_remove(unittest.TestCase):
+
+    def test_inplace(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                            theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+
+        z = Remove(True)(mySymbolicMatricesList, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myMatrix], z,
+                            accept_inplace=True)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x, y], y), [x]))
+
+    def test_sanity_check(self):
+        mySymbolicMatricesList = TypedListType(T.TensorType(
+                                theano.config.floatX, (False, False)))()
+        myMatrix = T.matrix()
+
+        z = Remove()(mySymbolicMatricesList, myMatrix)
+
+        f = theano.function([mySymbolicMatricesList, myMatrix], z)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        y = rand_ranged_matrix(-1000, 1000, [100, 101])
+
+        self.assertTrue(numpy.array_equal(f([x, y], y), [x]))

theano/typed_list/tests/test_type.py

+import unittest
+
+import numpy
+
+import theano
+import theano.typed_list
+from theano import tensor as T
+from theano.typed_list.type import TypedListType
+from theano.tests import unittest_tools as utt
+
+
+#took from tensors/tests/test_basic.py
+def rand_ranged_matrix(minimum, maximum, shape):
+    return numpy.asarray(numpy.random.rand(*shape) * (maximum - minimum)
+                         + minimum, dtype=theano.config.floatX)
+
+
+class test_typed_list_type(unittest.TestCase):
+
+    def setUp(self):
+        utt.seed_rng()
+
+    def test_wrong_input_on_creation(self):
+        """
+        Typed list type should raises an
+        error if the argument passed for
+        type is not a valid theano type
+        """
+
+        self.assertRaises(TypeError, TypedListType, None)
+
+    def test_wrong_input_on_filter(self):
+        """
+        Typed list type should raises an
+        error if the argument given to filter
+        isn't of the same type as the one
+        specified on creation
+        """
+
+        #list of matrices
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        self.assertRaises(TypeError, myType.filter, [4])
+
+    def test_not_a_list_on_filter(self):
+        """
+        Typed List Value should raises an error
+        if no iterable variable is given on input
+        """
+
+        #list of matrices
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        self.assertRaises(TypeError, myType.filter, 4)
+
+    def test_type_equality(self):
+        """
+        Typed list types should only be equal
+        when they contains the same theano
+        variables
+        """
+        #list of matrices
+        myType1 = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+        #list of matrices
+        myType2 = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+        #list of scalars
+        myType3 = TypedListType(T.TensorType(theano.config.floatX,
+                                            ()))
+
+        self.assertTrue(myType2 == myType1)
+        self.assertFalse(myType3 == myType1)
+
+    def test_filter_sanity_check(self):
+        """
+        Simple test on typed list type filter
+        """
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 100])
+
+        self.assertTrue(numpy.array_equal(myType.filter([x]), [x]))
+
+    def test_intern_filter(self):
+        """
+        Test checking if values contained are themselves
+        filtered. If they weren't this code would raise
+        an exception.
+        """
+        myType = TypedListType(T.TensorType('float64',
+                                            (False, False)))
+
+        x = numpy.asarray([[4, 5], [4, 5]], dtype='float32')
+
+        self.assertTrue(numpy.array_equal(myType.filter([x]), [x]))
+
+    #Will fail for unknown reasons
+    #under search
+    """
+    def test_load(self):
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 100])
+        testList = []
+        for i in range(10000):
+            testList.append(x)
+
+        self.assertTrue(numpy.array_equal(myType.filter(testList), testList))
+    """
+
+    def test_basic_nested_list(self):
+        """
+        Testing nested list with one level of depth
+        """
+        myNestedType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        myType = TypedListType(myNestedType)
+
+        x = rand_ranged_matrix(-1000, 1000, [100, 100])
+
+        self.assertTrue(numpy.array_equal(myType.filter([[x]]), [[x]]))
+
+    def test_comparison_different_depth(self):
+        """
+        Nested list with different depth aren't the same
+        """
+        myNestedType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        myNestedType2 = TypedListType(myNestedType)
+
+        myNestedType3 = TypedListType(myNestedType2)
+
+        self.assertFalse(myNestedType2 == myNestedType3)
+
+    def test_nested_list_arg(self):
+        """
+        test for the 'depth' optionnal argument
+        """
+        myNestedType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)), 3)
+
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        myManualNestedType = TypedListType(TypedListType(
+                             TypedListType(myType)))
+
+        self.assertTrue(myNestedType == myManualNestedType)
+
+    def test_get_depth(self):
+        """
+        test case for get_depth utilitary function
+        """
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        myManualNestedType = TypedListType(TypedListType(
+                             TypedListType(myType)))
+
+        self.assertTrue(myManualNestedType.get_depth() == 3)
+
+    def test_comparison_uneven_nested(self):
+        """
+        test for comparison between uneven nested list
+        """
+
+        myType = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))
+
+        myManualNestedType1 = TypedListType(TypedListType(
+                             TypedListType(myType)))
+
+        myManualNestedType2 = TypedListType(TypedListType(
+                             myType))
+
+        self.assertFalse(myManualNestedType1 == myManualNestedType2)
+        self.assertFalse(myManualNestedType2 == myManualNestedType1)
+
+    def test_variable_is_Typed_List_variable(self):
+        mySymbolicVariable = TypedListType(T.TensorType(theano.config.floatX,
+                                            (False, False)))()
+
+        self.assertTrue(isinstance(mySymbolicVariable,
+                                   theano.typed_list.TypedListVariable))

theano/typed_list/type.py

+from theano import gof
+
+
+class TypedListType(gof.Type):
+
+    def __init__(self, ttype, depth=0):
+        """
+        :Parameters:
+            -'ttype' : Type of theano variable this list
+            will contains, can be another list.
+            -'depth' : Optionnal parameters, any value
+            above 0 will create a nested list of this
+            depth. (0-based)
+        """
+        if depth < 0:
+            raise ValueError('Please specify a depth superior or'
+                            'equal to 0')
+        if not isinstance(ttype, gof.Type):
+            raise TypeError('Expected a Theano Type')
+
+        if depth == 0:
+            self.ttype = ttype
+        else:
+            self.ttype = TypedListType(ttype, depth - 1)
+
+    def filter(self, x, strict=False, allow_downcast=None):
+        """
+        :Parameters:
+            -'x' : value to filter
+            -'strict' : if true, only native python list will be accepted
+            -'allow_downcast' : does not have any utility at the moment
+        """
+        if strict:
+            if not isinstance(x, list):
+                raise TypeError('Expected a python list')
+        else:
+            x = [self.ttype.filter(y) for y in x]
+
+            if all(self.ttype.is_valid_value(y) for y in x):
+                return x
+
+            else:
+                raise TypeError('Expected all elements to'
+                                ' be %s' % str(self.ttype))
+
+    def __eq__(self, other):
+        """
+        two list are equals if they contains the same type.
+        """
+
+        return  type(self) == type(other) and self.ttype == other.ttype
+
+    def __hash__(self):
+        return gof.hashtype(self) ^ hash(self.ttype)
+
+    def __str__(self):
+        return 'TypedList <' + str(self.ttype) + '>'
+
+    def get_depth(self):
+        """
+        utilitary function to get the 0 based
+        level of the list
+        """
+        if isinstance(self.ttype, TypedListType):
+            return self.ttype.get_depth() + 1
+        else:
+            return 0