WIP soil

* Pandas integration
* Improved environment
* Logging and data dumps
* Tests
* Added Finite State Machine models
* Rewritten ipython notebook and documentation

tests/test_main.py

@@ -0,0 +1,225 @@
+from unittest import TestCase
+
+import os
+import yaml
+from functools import partial
+
+from os.path import join
+from soil import simulation, agents, utils
+
+
+ROOT = os.path.abspath(os.path.dirname(__file__))
+
+EXAMPLES = join(ROOT, '..', 'examples')
+
+
+class TestMain(TestCase):
+
+    def test_load_graph(self):
+        """
+        Load a graph from file if the extension is known.
+        Raise an exception otherwise.
+        """
+        config = {
+            'network_params': {
+                'path': join(ROOT, 'test.gexf')
+            }
+        }
+        G = utils.load_network(config['network_params'])
+        assert G
+        assert len(G) == 2
+        with self.assertRaises(AttributeError):
+            config = {
+                'network_params': {
+                    'path': join(ROOT, 'unknown.extension')
+                }
+            }
+            G = utils.load_network(config['network_params'])
+            print(G)
+
+    def test_generate_barabasi(self):
+        """
+        If no path is given, a generator and network parameters
+        should be used to generate a network
+        """
+        config = {
+            'network_params': {
+                'generator': 'barabasi_albert_graph'
+            }
+        }
+        with self.assertRaises(TypeError):
+            G = utils.load_network(config['network_params'])
+        config['network_params']['n'] = 100
+        config['network_params']['m'] = 10
+        G = utils.load_network(config['network_params'])
+        assert len(G) == 100
+
+    def test_empty_simulation(self):
+        """A simulation with a base behaviour should do nothing"""
+        config = {
+            'network_params': {
+                'path': join(ROOT, 'test.gexf')
+            },
+            'agent_type': 'NetworkAgent',
+            'environment_params': {
+            }
+        }
+        s = simulation.from_config(config)
+        s.run_simulation()
+
+    def test_counter_agent(self):
+        """
+        The initial states should be applied to the agent and the
+        agent should be able to update its state."""
+        config = {
+            'network_params': {
+                'path': join(ROOT, 'test.gexf')
+            },
+            'agent_type': 'CounterModel',
+            'states': [{'neighbors': 10}, {'total': 12}],
+            'max_time': 2,
+            'num_trials': 1,
+            'environment_params': {
+            }
+        }
+        s = simulation.from_config(config)
+        env = s.run_simulation()[0]
+        assert env.get_agent(0)['neighbors', 0] == 10
+        assert env.get_agent(0)['neighbors', 1] == 1
+        assert env.get_agent(1)['total', 0] == 12
+        assert env.get_agent(1)['neighbors', 1] == 1
+
+    def test_counter_agent_history(self):
+        """
+        The evolution of the state should be recorded in the logging agent
+        """
+        config = {
+            'network_params': {
+                'path': join(ROOT, 'test.gexf')
+            },
+            'network_agents': [{
+                'agent_type': 'AggregatedCounter',
+                'weight': 1,
+                'state': {'id': 0}
+
+            }],
+            'max_time': 10,
+            'environment_params': {
+            }
+        }
+        s = simulation.from_config(config)
+        env = s.run_simulation()[0]
+        for agent in env.network_agents:
+            last = 0
+            assert len(agent._history) == 11
+            for step, total in agent['total', None].items():
+                if step > 0:
+                    assert total == last + 2
+                    last = total
+
+    def test_custom_agent(self):
+        """Allow for search of neighbors with a certain state_id"""
+        class CustomAgent(agents.NetworkAgent):
+            def step(self):
+                self.state['neighbors'] = self.count_agents(state_id=0,
+                                                            limit_neighbors=True)
+        config = {
+            'network_params': {
+                'path': join(ROOT, 'test.gexf')
+            },
+            'network_agents': [{
+                'agent_type': CustomAgent,
+                'weight': 1,
+                'state': {'id': 0}
+
+            }],
+            'max_time': 10,
+            'environment_params': {
+            }
+        }
+        s = simulation.from_config(config)
+        env = s.run_simulation()[0]
+        assert env.get_agent(0).state['neighbors'] == 1
+
+    def test_torvalds_example(self):
+        """A complete example from a documentation should work."""
+        config = utils.load_file(join(EXAMPLES, 'torvalds.yml'))[0]
+        config['network_params']['path'] = join(EXAMPLES,
+                                                config['network_params']['path'])
+        s = simulation.from_config(config)
+        env = s.run_simulation()[0]
+        for a in env.network_agents:
+            skill_level = a.state['skill_level']
+            if a.id == 'Torvalds':
+                assert skill_level == 'God'
+                assert a.state['total'] == 3
+                assert a.state['neighbors'] == 2
+            elif a.id == 'balkian':
+                assert skill_level == 'developer'
+                assert a.state['total'] == 3
+                assert a.state['neighbors'] == 1
+            else:
+                assert skill_level == 'beginner'
+                assert a.state['total'] == 3
+                assert a.state['neighbors'] == 1
+
+    def test_yaml(self):
+        """
+        The YAML version of a newly created simulation
+        should be equivalent to the configuration file used
+        """
+        with utils.timer('loading'):
+            config = utils.load_file(join(EXAMPLES, 'complete.yml'))[0]
+            s = simulation.from_config(config)
+        with utils.timer('serializing'):
+            serial = s.to_yaml()
+        with utils.timer('recovering'):
+            recovered = yaml.load(serial)
+        with utils.timer('deleting'):
+            del recovered['topology']
+        assert config == recovered
+
+    def test_configuration_changes(self):
+        """
+        The configuration should not change after running
+         the simulation.
+        """
+        config = utils.load_file('examples/complete.yml')[0]
+        s = simulation.from_config(config)
+        for i in range(5):
+            s.run_simulation()
+            nconfig = s.to_dict()
+            del nconfig['topology']
+            assert config == nconfig
+
+    def test_examples(self):
+        """
+        Make sure all examples in the examples folder are correct
+        """
+
+
+def make_example_test(path, config):
+    def wrapped(self):
+        root = os.getcwd()
+        os.chdir(os.path.dirname(path))
+        s = simulation.from_config(config)
+        envs = s.run_simulation()
+        for env in envs:
+            n = config['network_params'].get('n', None)
+            if n is not None:
+                assert len(env.get_agents()) == n
+        os.chdir(root)
+    return wrapped
+
+
+def add_example_tests():
+    for config, path in utils.load_config(join(EXAMPLES, '*.yml')):
+        p = make_example_test(path=path, config=config)
+        fname = os.path.basename(path)
+        p.__name__ = 'test_example_file_%s' % fname
+        p.__doc__ = '%s should be a valid configuration' % fname
+        setattr(TestMain, p.__name__, p)
+        del p
+
+
+add_example_tests()

tests/unknown.extension

@@ -0,0 +1 @@
+This is an unknown file