Dirty Parallelize Simulations

engine/mechanismExecutor.py

@@ -1,7 +1,7 @@
 from copy import deepcopy
 from fn import _
 from fn.op import foldr, call
-from ui.config import behavior_ops
+from ui.config2 import behavior_ops
 
 
 def getColResults(step, sL, s, funcs):

engine/multiproc.py

@@ -2,10 +2,7 @@ from pathos.multiprocessing import ProcessingPool as Pool
 
 
 def parallelize_simulations(f, states_list, configs, env_processes, T, N):
-    def process(config):
-        return f(states_list, config, env_processes, T, N)
-
     with Pool(len(configs)) as p:
-        results = p.map(process, configs)
+        results = p.map(lambda x: f(states_list, x[0], x[1], T, N), list(zip(configs, env_processes)))
 
     return results

engine/run.py

@@ -4,26 +4,53 @@ from tabulate import tabulate
 from engine.configProcessor import generate_config, create_tensor_field
 from engine.mechanismExecutor import simulation
 from engine.utils import flatten
-from ui.config import state_dict, mechanisms, exogenous_states, env_processes, sim_config
 from engine.multiproc import parallelize_simulations
 
+from decimal import Decimal
+
+# from ui.config import state_dict, mechanisms, exogenous_states, env_processes, sim_config
+
+import ui.config1 as conf1
+import ui.config2 as conf2
 
 def main():
-    states_list = [state_dict]
-    ep = list(exogenous_states.values())
-    config = generate_config(state_dict, mechanisms, ep)
+    state_dict = {
+        's1': Decimal(0.0),
+        's2': Decimal(0.0),
+        's3': Decimal(1.0),
+        's4': Decimal(1.0),
+        'timestamp': '2018-10-01 15:16:24'
+    }
+    sim_config = {
+        "N": 2,
+        "T": range(5)
+    }
+
     T = sim_config['T']
     N = sim_config['N']
-    configs = [config, config]
+    states_list = [state_dict]
+
+    ep1 = list(conf1.exogenous_states.values())
+    ep2 = list(conf2.exogenous_states.values())
+    eps = [ep1,ep2]
+
+    config1 = generate_config(conf1.state_dict, conf1.mechanisms, ep1)
+    config2 = generate_config(conf2.state_dict, conf2.mechanisms, ep2)
+
+    mechanisms = [conf1.mechanisms, conf2.mechanisms]
+
+    configs = [config1, config2]
+    env_processes = [conf1.env_processes, conf2.env_processes]
 
     # Dimensions: N x r x mechs
 
     if len(configs) > 1:
-     simulations = parallelize_simulations(simulation, states_list, configs, env_processes, T, N)
-    else:
-     simulations = [simulation(states_list, configs[0], env_processes, T, N)]
+        simulations = parallelize_simulations(simulation, states_list, configs, env_processes, T, N)
+    # else:
+    #     simulations = [simulation(states_list, configs[0], env_processes, T, N)]
 
-    for result in simulations:
-        print(tabulate(create_tensor_field(mechanisms, ep), headers='keys', tablefmt='psql'))
-        print
+    # simulations = [simulation(states_list, config1, conf1.env_processes, T, N)]
+
+    for result, mechanism, ep in list(zip(simulations, mechanisms, eps)):
+        print(tabulate(create_tensor_field(mechanism, ep), headers='keys', tablefmt='psql'))
         print(tabulate(pd.DataFrame(flatten(result)), headers='keys', tablefmt='psql'))

notebooks/test.ipynb

@@ -10,13 +10,13 @@
     {
      "ename": "ImportError",
      "evalue": "cannot import name 'run'",
-     "output_type": "error",
      "traceback": [
       "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
       "\u001b[1;31mImportError\u001b[0m                               Traceback (most recent call last)",
       "\u001b[1;32m<ipython-input-5-a6e895c51fc0>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[1;32mfrom\u001b[0m \u001b[0mengine\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0mrun\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      2\u001b[0m \u001b[0mrun\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmain\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
       "\u001b[1;31mImportError\u001b[0m: cannot import name 'run'"
-     ]
+     ],
+     "output_type": "error"
     }
    ],
    "source": [

ui/config1.py

@@ -0,0 +1,212 @@
+from engine.utils import bound_norm_random, ep_time_step, proc_trigger, exo_update_per_ts
+from fn.op import foldr
+from fn import _
+from fn.func import curried
+
+import numpy as np
+from decimal import Decimal
+
+seed = {
+    'z': np.random.RandomState(1),
+    'a': np.random.RandomState(2),
+    'b': np.random.RandomState(3),
+    'c': np.random.RandomState(3)
+}
+
+# # Behaviors per Mechanism
+# def b1m1(step, sL, s):
+#     return np.array([1, 2])
+# def b2m1(step, sL, s):
+#     return np.array([3, 4])
+# # Internal States per Mechanism
+# def s1m1(step, sL, s, _input):
+#     y = 's1'
+#     x = _input['b1'] * s['s1'] + _input['b2']
+#     return (y, x)
+
+# Behaviors per Mechanism
+# Different return types per mechanism ?? *** No ***
+def b1m1(step, sL, s):
+    return {'param1': 1}
+def b2m1(step, sL, s):
+    return {'param2': 4}
+
+def b1m2(step, sL, s):
+    return {'param1': 'a', 'param2': 2}
+def b2m2(step, sL, s):
+    return {'param1': 'b', 'param2': 4}
+
+def b1m3(step, sL, s):
+    return {'param1': ['c'], 'param2': np.array([10, 100])}
+def b2m3(step, sL, s):
+    return {'param1': ['d'], 'param2': np.array([20, 200])}
+
+
+# Internal States per Mechanism
+def s1m1(step, sL, s, _input):
+    y = 's1'
+    x = _input['param1']
+    return (y, x)
+def s2m1(step, sL, s, _input):
+    y = 's2'
+    x = _input['param2']
+    return (y, x)
+
+def s1m2(step, sL, s, _input):
+    y = 's1'
+    x = _input['param1']
+    return (y, x)
+def s2m2(step, sL, s, _input):
+    y = 's2'
+    x = _input['param2']
+    return (y, x)
+
+def s1m3(step, sL, s, _input):
+    y = 's1'
+    x = _input['param1']
+    return (y, x)
+def s2m3(step, sL, s, _input):
+    y = 's2'
+    x = _input['param2']
+    return (y, x)
+
+# Exogenous States
+proc_one_coef_A = 0.7
+proc_one_coef_B = 1.3
+
+def es3p1(step, sL, s, _input):
+    y = 's3'
+    x = s['s3'] * bound_norm_random(seed['a'], proc_one_coef_A, proc_one_coef_B)
+    return (y, x)
+
+def es4p2(step, sL, s, _input):
+    y = 's4'
+    x = s['s4'] * bound_norm_random(seed['b'], proc_one_coef_A, proc_one_coef_B)
+    return (y, x)
+
+def es5p2(step, sL, s, _input): # accept timedelta instead of timedelta params
+    y = 'timestamp'
+    x = ep_time_step(s, s['timestamp'], seconds=1)
+    return (y, x)
+
+
+# Environment States
+def env_a(x):
+    return 10
+def env_b(x):
+    return 10
+# def what_ever(x):
+#     return x + 1
+
+# Genesis States
+state_dict = {
+    's1': Decimal(0.0),
+    's2': Decimal(0.0),
+    's3': Decimal(1.0),
+    's4': Decimal(1.0),
+    'timestamp': '2018-10-01 15:16:24'
+}
+
+# remove `exo_update_per_ts` to update every ts
+exogenous_states = exo_update_per_ts(
+    {
+    "s3": es3p1,
+    "s4": es4p2,
+    "timestamp": es5p2
+    }
+)
+
+#	make env proc trigger field agnostic
+env_processes = {
+    "s3": proc_trigger('2018-10-01 15:16:25', env_a),
+    "s4": proc_trigger('2018-10-01 15:16:25', env_b)
+}
+
+# lambdas
+# genesis Sites should always be there
+# [1, 2]
+# behavior_ops = [ foldr(_ + _), lambda x: x + 0 ]
+def print_fwd(x):
+    print(x)
+    return x
+
+def behavior_to_dict(v):
+    return dict(list(zip(map(lambda n: 'b' + str(n+1), list(range(len(v)))), v)))
+
+@curried
+def foldr_dict_vals(f, d):
+    return foldr(f)(list(d.values()))
+
+def sum_dict_values():
+    return foldr_dict_vals(_ + _)
+
+def get_base_value(datatype):
+    if datatype is str:
+        return ''
+    elif datatype is int:
+        return 0
+    elif datatype is list:
+        return []
+    return 0
+
+
+@curried
+def dict_op(f, d1, d2):
+
+    def set_base_value(target_dict, source_dict, key):
+        if key not in target_dict:
+            return get_base_value(type(source_dict[key]))
+        else:
+            return target_dict[key]
+
+    key_set = set(list(d1.keys())+list(d2.keys()))
+
+    return {k: f(set_base_value(d1, d2, k), set_base_value(d2, d1, k)) for k in key_set}
+
+def dict_elemwise_sum():
+    return dict_op(_ + _)
+
+# [1, 2] = {'b1': ['a'], 'b2', [1]} =
+# behavior_ops = [ behavior_to_dict, print_fwd, sum_dict_values ]
+behavior_ops = [ foldr(dict_elemwise_sum()) ]
+# behavior_ops = []
+
+# need at least 1 behaviour and 1 state function for the 1st mech with behaviors
+# mechanisms = {}
+mechanisms = {
+    "m1": {
+        "behaviors": {
+            "b1": b1m1, # lambda step, sL, s: s['s1'] + 1,
+            "b2": b2m1
+        },
+        "states": { # exclude only. TypeError: reduce() of empty sequence with no initial value
+            "s1": s1m1,
+            "s2": s2m1
+        }
+    },
+    "m2": {
+        "behaviors": {
+            "b1": b1m2,
+            "b2": b2m2
+        },
+        "states": {
+            "s1": s1m2,
+            "s2": s2m2
+        }
+    },
+    "m3": {
+        "behaviors": {
+            "b1": b1m3,
+            "b2": b2m3
+        },
+        "states": {
+            "s1": s1m3,
+            "s2": s2m3
+        }
+    }
+}
+
+sim_config = {
+    "N": 2,
+    "T": range(5)
+}