Moved graph algorithms into a separate module.

Validate activation function signatures.

Author: CodeReclaimers

neat/activations.py

@@ -1,3 +1,4 @@
+import inspect
 import math
 
 
@@ -22,7 +23,7 @@ def gauss_activation(z):
 
 
 def relu_activation(z):
-    return z if z > 0.0 else 0
+    return z if z > 0.0 else 0.0
 
 
 def identity_activation(z):
@@ -70,6 +71,15 @@ class InvalidActivationFunction(Exception):
     pass
 
 
+def validate_activation(function):
+    if not inspect.isfunction(function):
+        raise InvalidActivationFunction("A function object is required.")
+
+    args = inspect.getargspec(function.__call__)
+    if len(args[0]) != 1:
+        raise InvalidActivationFunction("A single-argument function is required.")
+
+
 class ActivationFunctionSet(object):
     def __init__(self):
         self.functions = {'sigmoid': sigmoid_activation,
@@ -88,7 +98,7 @@ def __init__(self):
                           'cube': cube_activation}
 
     def add(self, config_name, function):
-        # TODO: Verify that the given function has the correct signature.
+        validate_activation(function)
         self.functions[config_name] = function
 
     def get(self, config_name):
@@ -99,7 +109,4 @@ def get(self, config_name):
         return f
 
     def is_valid(self, config_name):
-        # TODO: Verify that the given function has the correct signature.
         return config_name in self.functions
-
-

neat/graphs.py

@@ -0,0 +1,93 @@
+""" Directed graph algorithm implementations. """
+
+
+def creates_cycle(connections, test):
+    """
+    Returns true if the addition of the "test" connection would create a cycle,
+    assuming that no cycle already exists in the graph represented by "connections".
+    """
+    i, o = test
+    if i == o:
+        return True
+
+    visited = set([o])
+    while True:
+        num_added = 0
+        for a, b in connections:
+            if a in visited and b not in visited:
+                if b == i:
+                    return True
+
+                visited.add(b)
+                num_added += 1
+
+        if num_added == 0:
+            return False
+
+
+def required_for_output(inputs, outputs, connections):
+    '''
+    Collect the nodes whose state is required to compute the final network output(s).
+    :param inputs: list of the input identifiers
+    :param outputs: list of the output node identifiers
+    :param connections: list of (input, output) connections in the network.
+    NOTE: It is assumed that the input identifier set and the node identifier set are disjoint.
+    By convention, the output node ids are always the same as the output index.
+
+    Returns a list of layers, with each layer consisting of a set of identifiers.
+    '''
+
+    required = set(outputs)
+    S = set(outputs)
+    while 1:
+        # Find nodes not in S whose output is consumed by a node in S.
+        T = set(a for (a, b) in connections if b in S and a not in S)
+
+        if not T:
+            break
+
+        layer_nodes = set(x for x in T if x not in inputs)
+        if not layer_nodes:
+            break
+
+        required = required.union(layer_nodes)
+        S = S.union(T)
+
+    return required
+
+
+def feed_forward_layers(inputs, outputs, connections):
+    '''
+    Collect the layers whose members can be evaluated in parallel in a feed-forward network.
+    :param inputs: list of the network input nodes
+    :param outputs: list of the output node identifiers
+    :param connections: list of (input, output) connections in the network.
+
+    Returns a list of layers, with each layer consisting of a set of node identifiers.
+    Note that the returned layers do not contain nodes whose output is ultimately
+    never used to compute the final network output.
+    '''
+
+    required = required_for_output(inputs, outputs, connections)
+
+    layers = []
+    S = set(inputs)
+    while 1:
+        # Find candidate nodes C for the next layer.  These nodes should connect
+        # a node in S to a node not in S.
+        C = set(b for (a, b) in connections if a in S and b not in S)
+        # Keep only the used nodes whose entire input set is contained in S.
+        T = set()
+        for n in C:
+            if n in required and all(a in S for (a, b) in connections if b == n):
+                T.add(n)
+
+        if not T:
+            break
+
+        layers.append(T)
+        S = S.union(T)
+
+    return layers
+
+

neat/nn/__init__.py

@@ -1,97 +1,6 @@
+from neat.graphs import creates_cycle, required_for_output, feed_forward_layers
 from neat.six_util import iterkeys, itervalues, iteritems
 
-# TODO: All this directed graph logic should be in a core module.
-
-
-def creates_cycle(connections, test):
-    """
-    Returns true if the addition of the "test" connection would create a cycle,
-    assuming that no cycle already exists in the graph represented by "connections".
-    """
-    i, o = test
-    if i == o:
-        return True
-
-    visited = set([o])
-    while True:
-        num_added = 0
-        for a, b in connections:
-            if a in visited and b not in visited:
-                if b == i:
-                    return True
-
-                visited.add(b)
-                num_added += 1
-
-        if num_added == 0:
-            return False
-
-
-def required_for_output(inputs, outputs, connections):
-    '''
-    Collect the nodes whose state is required to compute the final network output(s).
-    :param inputs: list of the input identifiers
-    :param outputs: list of the output node identifiers
-    :param connections: list of (input, output) connections in the network.
-    NOTE: It is assumed that the input identifier set and the node identifier set are disjoint.
-    By convention, the output node ids are always the same as the output index.
-
-    Returns a list of layers, with each layer consisting of a set of identifiers.
-    '''
-
-    required = set(outputs)
-    S = set(outputs)
-    while 1:
-        # Find nodes not in S whose output is consumed by a node in S.
-        T = set(a for (a, b) in connections if b in S and a not in S)
-
-        if not T:
-            break
-
-        layer_nodes = set(x for x in T if x not in inputs)
-        if not layer_nodes:
-            break
-
-        required = required.union(layer_nodes)
-        S = S.union(T)
-
-    return required
-
-
-def feed_forward_layers(inputs, outputs, connections):
-    '''
-    Collect the layers whose members can be evaluated in parallel in a feed-forward network.
-    :param inputs: list of the network input nodes
-    :param outputs: list of the output node identifiers
-    :param connections: list of (input, output) connections in the network.
-
-    Returns a list of layers, with each layer consisting of a set of node identifiers.
-    Note that the returned layers do not contain nodes whose output is ultimately
-    never used to compute the final network output.
-    '''
-
-    required = required_for_output(inputs, outputs, connections)
-
-    layers = []
-    S = set(inputs)
-    while 1:
-        # Find candidate nodes C for the next layer.  These nodes should connect
-        # a node in S to a node not in S.
-        C = set(b for (a, b) in connections if a in S and b not in S)
-        # Keep only the used nodes whose entire input set is contained in S.
-        T = set()
-        for n in C:
-            if n in required and all(a in S for (a, b) in connections if b == n):
-                T.add(n)
-
-        if not T:
-            break
-
-        layers.append(T)
-        S = S.union(T)
-
-    return layers
-
 
 class FeedForwardNetwork(object):
     def __init__(self, max_node, inputs, outputs, node_evals):

tests/test_feedforward.py

@@ -1,157 +1,11 @@
 import random
-
 from neat import nn
 
 
 def assert_almost_equal(x, y, tol):
     assert abs(x - y) < tol, "{!r} !~= {!r}".format(x, y)
 
 
-def test_creates_cycle():
-    assert nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (0, 0))
-
-    assert nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (1, 0))
-    assert not nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (0, 1))
-
-    assert nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (2, 0))
-    assert not nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (0, 2))
-
-    assert nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (3, 0))
-    assert not nn.creates_cycle([(0, 1), (1, 2), (2, 3)], (0, 3))
-
-    assert nn.creates_cycle([(0, 2), (1, 3), (2, 3), (4, 2)], (3, 4))
-    assert not nn.creates_cycle([(0, 2), (1, 3), (2, 3), (4, 2)], (4, 3))
-
-
-def test_required_for_output():
-    inputs = [0, 1]
-    outputs = [2]
-    connections = [(0, 2), (1, 2)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2} == required
-
-    inputs = [0, 1]
-    outputs = [2]
-    connections = [(0, 3), (1, 4), (3, 2), (4, 2)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2, 3, 4} == required
-
-    inputs = [0, 1]
-    outputs = [3]
-    connections = [(0, 2), (1, 2), (2, 3)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2, 3} == required
-
-    inputs = [0, 1]
-    outputs = [4]
-    connections = [(0, 2), (1, 2), (1, 3), (2, 3), (2, 4), (3, 4)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2, 3, 4} == required
-
-    inputs = [0, 1]
-    outputs = [4]
-    connections = [(0, 2), (1, 3), (2, 3), (3, 4), (4, 2)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2, 3, 4} == required
-
-    inputs = [0, 1]
-    outputs = [4]
-    connections = [(0, 2), (1, 2), (1, 3), (2, 3), (2, 4), (3, 4), (2, 5)]
-    required = nn.required_for_output(inputs, outputs, connections)
-    assert {2, 3, 4} == required
-
-
-def test_fuzz_required():
-    for _ in range(1000):
-        n_hidden = random.randint(10, 100)
-        n_in = random.randint(1, 10)
-        n_out = random.randint(1, 10)
-        nodes = list(set(random.randint(0, 1000) for _ in range(n_in + n_out + n_hidden)))
-        random.shuffle(nodes)
-
-        inputs = nodes[:n_in]
-        outputs = nodes[n_in:n_in + n_out]
-        connections = []
-        for _ in range(n_hidden * 2):
-            a = random.choice(nodes)
-            b = random.choice(nodes)
-            if a == b:
-                continue
-            if a in inputs and b in inputs:
-                continue
-            if a in outputs and b in outputs:
-                continue
-            connections.append((a, b))
-
-        required = nn.required_for_output(inputs, outputs, connections)
-        for o in outputs:
-            assert o in required
-
-
-def test_feed_forward_layers():
-    inputs = [0, 1]
-    outputs = [2]
-    connections = [(0, 2), (1, 2)]
-    layers = nn.feed_forward_layers(inputs, outputs, connections)
-    assert [{2}] == layers
-
-    inputs = [0, 1]
-    outputs = [3]
-    connections = [(0, 2), (1, 2), (2, 3)]
-    layers = nn.feed_forward_layers(inputs, outputs, connections)
-    assert [{2}, {3}] == layers
-
-    inputs = [0, 1]
-    outputs = [4]
-    connections = [(0, 2), (1, 2), (1, 3), (2, 3), (2, 4), (3, 4)]
-    layers = nn.feed_forward_layers(inputs, outputs, connections)
-    assert [{2}, {3}, {4}] == layers
-
-    inputs = [0, 1, 2, 3]
-    outputs = [11, 12, 13]
-    connections = [(0, 4), (1, 4), (1, 5), (2, 5), (2, 6), (3, 6), (3, 7),
-                   (4, 8), (5, 8), (5, 9), (5, 10), (6, 10), (6, 7),
-                   (8, 11), (8, 12), (8, 9), (9, 10), (7, 10),
-                   (10, 12), (10, 13)]
-    layers = nn.feed_forward_layers(inputs, outputs, connections)
-    assert [{4, 5, 6}, {8, 7}, {9, 11}, {10}, {12, 13}] == layers
-
-    inputs = [0, 1, 2, 3]
-    outputs = [11, 12, 13]
-    connections = [(0, 4), (1, 4), (1, 5), (2, 5), (2, 6), (3, 6), (3, 7),
-                   (4, 8), (5, 8), (5, 9), (5, 10), (6, 10), (6, 7),
-                   (8, 11), (8, 12), (8, 9), (9, 10), (7, 10),
-                   (10, 12), (10, 13),
-                   (3, 14), (14, 15), (5, 16), (10, 16)]
-    layers = nn.feed_forward_layers(inputs, outputs, connections)
-    assert [{4, 5, 6}, {8, 7}, {9, 11}, {10}, {12, 13}] == layers
-
-
-def test_fuzz_feed_forward_layers():
-    for _ in range(1000):
-        n_hidden = random.randint(10, 100)
-        n_in = random.randint(1, 10)
-        n_out = random.randint(1, 10)
-        nodes = list(set(random.randint(0, 1000) for _ in range(n_in + n_out + n_hidden)))
-        random.shuffle(nodes)
-
-        inputs = nodes[:n_in]
-        outputs = nodes[n_in:n_in + n_out]
-        connections = []
-        for _ in range(n_hidden * 2):
-            a = random.choice(nodes)
-            b = random.choice(nodes)
-            if a == b:
-                continue
-            if a in inputs and b in inputs:
-                continue
-            if a in outputs and b in outputs:
-                continue
-            connections.append((a, b))
-
-        nn.feed_forward_layers(inputs, outputs, connections)
-
-
 # TODO: Update this test for the current implementation.
 # def test_simple_nohidden():
 #     config_params = {