initial commit

Author: mpezeshki

README.md

@@ -1,2 +1,15 @@
 # Theano_Tile_Coding
 A tile coder in theano for Reinforcement Learning tasks
+
+## An example of 2D function approximation
+![](https://github.com/mohammadpz/Theano_Tile_Coding/blob/master/example.gif)
+
+
+## Disclaimer
+THE CODE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+PERFORMANCE OF THIS CODE.

example.gif

@@ -0,0 +1,80 @@
+import numpy as np
+import theano
+import theano.tensor as T
+
+
+class TileCodingLayer:
+    def __init__(self, min_val, max_val, num_tiles,
+                 num_tilings, num_features):
+        self.min_val = min_val
+        self.max_val = max_val
+        self.num_tiles = num_tiles
+        self.num_tilings = num_tilings
+        self.num_features = num_features
+
+        # The range of each feature is divided into num_tile tiles.
+        # But we keep one more tile. The reason is that since we shift
+        # tilings by an offset, we want to make sure that every point
+        # after shifting maps to one tile.
+        # Shape: (num_tilings, (num_tiles+1) * num_features)
+        self.weight = theano.shared(np.zeros(
+            (num_tilings,) + (num_tiles + 1,) * num_features))
+
+    def quantize(self, x):
+        len_tile = float(self.max_val - self.min_val) / self.num_tiles
+        offsets = (len_tile * T.arange(self.num_tilings).astype('float32') /
+                   self.num_tilings)
+        # shapes: (features, num_tilings) = (features, 1) + (1, num_tilings)
+        mapped_x = x.dimshuffle(0, 'x') + offsets.dimshuffle('x', 0)
+        # Since we have an extra tile:
+        new_max_val = self.max_val + len_tile
+        # Mapping into range [0, num_tiles + 1] to be able to do quantization
+        # by casting to int.
+        mapped_x = ((self.num_tiles + 1) * mapped_x /
+                    (new_max_val - self.min_val))
+        # shape: (num_tilings, num_features)
+        q_x = mapped_x.astype('int32').T
+        return q_x
+
+    def approximate(self, q_x):
+        indices = [T.arange(self.num_tilings)]
+        for f in range(self.num_features):
+            indices += [q_x[:, f]]
+        self.indices = indices
+        # shape: (num_tilings,)
+        selected = self.weight.__getitem__(tuple(indices))
+        y_hat = T.sum(selected)
+        return y_hat
+
+    def update_rule(self, y, y_hat, learning_rate):
+        # grad is only used to locate the weights which
+        # are used for computing y_hat. Since
+        # "y_hat = T.sum(selected)", T.grad will returns
+        # 1 for corresponding weights and 0 for others.
+        grad_ = T.grad(y_hat, self.weight)
+        learning_rate = learning_rate / self.num_tilings
+        upd = grad_ * learning_rate * (y - y_hat)
+        updates = [(self.weight, self.weight + upd)]
+        return updates
+
+
+def get_tile_coder(min_val, max_val, num_tiles, num_tilings,
+                   num_features, learning_rate):
+    # x.shape: (num_features), y.shape: ()
+    x = T.fvector('x')
+    y = T.fscalar('y')
+
+    tile_coding_layer = TileCodingLayer(
+        min_val=min_val, max_val=max_val,
+        num_tiles=num_tiles, num_tilings=num_tilings,
+        num_features=num_features)
+
+    # quantized_x
+    q_x = tile_coding_layer.quantize(x)
+    y_hat = tile_coding_layer.approximate(q_x)
+    updates = tile_coding_layer.update_rule(y, y_hat, 0.1)
+
+    train = theano.function([x, y], y_hat, updates=updates, allow_input_downcast=True)
+    eval_ = theano.function([x], y_hat, allow_input_downcast=True)
+
+    return train, eval_

layers.py

@@ -0,0 +1,77 @@
+import numpy as np
+import matplotlib.pyplot as plt
+from matplotlib import cm
+from mpl_toolkits.mplot3d import Axes3D
+from layers import get_tile_coder
+
+
+# x.shape: (num_features)
+def target_function(x):
+    return np.sin(x[0]) + np.cos(x[1]) + 0.01 * np.random.randn()
+
+
+def get_dataset(num_samples, min_val, max_val):
+    # num_features is assumed to be 2 for this example.
+    # However it can take any value in general.
+    dataset = []
+    for i in range(num_samples):
+        x = np.array([np.random.random() * 7, np.random.random() * 7])
+        y = target_function(x)
+        dataset.append((x, y))
+    return dataset
+
+
+def plot_function(ax, function, text, hold=False):
+    ax.cla()
+    x_0 = np.linspace(0, 7, 100)
+    x_1 = np.linspace(0, 7, 100)
+    z = np.zeros((100, 100))
+    for i in range(100):
+        for j in range(100):
+            z[j, i] = function(np.array([x_0[i], x_1[j]]))
+    X_0, X_1 = np.meshgrid(x_0, x_1)
+    ax.plot_surface(X_0, X_1, z, rstride=8, cstride=8, alpha=0.3)
+    ax.contourf(X_0, X_1, z, zdir='z', offset=-3, cmap=cm.coolwarm)
+    ax.contourf(X_0, X_1, z, zdir='x', offset=-1, cmap=cm.coolwarm)
+    ax.contourf(X_0, X_1, z, zdir='y', offset=-1, cmap=cm.coolwarm)
+    ax.set_xlim(-1, 8)
+    ax.set_ylim(-1, 8)
+    ax.set_zlim(-3, 3)
+    ax.view_init(45, 45)
+    ax.set_title(text)
+    if hold:
+        plt.show()
+    else:
+        plt.draw()
+        plt.savefig(text + '.png')
+        plt.pause(.0001)
+
+train, eval_ = get_tile_coder(
+    min_val=0, max_val=7.0, num_tiles=10,
+    num_tilings=10, num_features=2,
+    learning_rate=0.1)
+
+dataset = get_dataset(10000, 0, 7)
+fig = plt.figure(figsize=np.array([12, 5]))
+ax_0 = fig.add_subplot(1, 2, 1, projection='3d')
+ax_1 = fig.add_subplot(1, 2, 2, projection='3d')
+plot_function(ax_0, target_function, 'Target function')
+# import ipdb; ipdb.set_trace()
+
+print 'Training'
+for i, datapoint in enumerate(dataset):
+    x, y = datapoint
+    train(x.astype('float32'), y.astype('float32'))
+    if i <= 20:
+        step = 5
+    elif i <= 100:
+        step = 20
+    elif i <= 1500:
+        step = 100
+    else:
+        step = 1000
+    if i % step == 0:
+        plot_function(ax_1, eval_, 'Seen points: ' + str(i))
+
+    if i == len(dataset) - 1:
+        plot_function(ax_1, eval_, 'Learned Function', True)