change hints for 3.9 compat

cadCAD/configuration/__init__.py

@@ -179,7 +179,7 @@ def append_model(
 
 
 class Identity:
-    def __init__(self, policy_id: dict[str, int] = {'identity': 0}) -> None:
+    def __init__(self, policy_id: Dict[str, int] = {'identity': 0}) -> None:
         self.beh_id_return_val = policy_id
 
     def p_identity(self, var_dict, sub_step, sL, s, **kwargs):
@@ -199,7 +199,7 @@ def state_identity(self, k: str) -> callable:
     def apply_identity_funcs(self,
                              identity: callable,
                              df: DataFrame,
-                             cols: list[str]) -> DataFrame:
+                             cols: List[str]) -> DataFrame:
         """
         Apply the identity on each df column, using its self value as the
         argument.
@@ -232,7 +232,7 @@ def create_matrix_field(self, partial_state_updates, key: str) -> DataFrame:
             return pd.DataFrame({'empty': []})
 
     def generate_config(self, initial_state, partial_state_updates, exo_proc
-                       ) -> list[tuple[list[callable], list[callable]]]:
+                       ) -> List[tuple[list[callable], List[callable]]]:
 
         def no_update_handler(bdf, sdf):
             if (bdf.empty == False) and (sdf.empty == True):

cadCAD/engine/__init__.py

@@ -10,10 +10,10 @@
 from cadCAD.engine.execution import single_proc_exec, parallelize_simulations, local_simulations
 from cadCAD.types import *
 
-VarDictType = dict[str, list[object]]
-StatesListsType = list[dict[str, object]]
-ConfigsType = list[tuple[list[callable], list[callable]]]
-EnvProcessesType = dict[str, callable]
+VarDictType = Dict[str, List[object]]
+StatesListsType = List[dict[str, object]]
+ConfigsType = List[tuple[list[callable], List[callable]]]
+EnvProcessesType = Dict[str, callable]
 
 
 class ExecutionMode:
@@ -80,7 +80,7 @@ def __init__(self,
         self.configs = configs
         self.empty_return = empty_return
 
-    def execute(self) -> tuple[object, object, dict[str, object]]:
+    def execute(self) -> Tuple[object, object, Dict[str, object]]:
         if self.empty_return is True:
             return [], [], []
 

cadCAD/engine/execution.py

@@ -4,10 +4,10 @@
 from cadCAD.types import *
 from cadCAD.utils import flatten
 
-VarDictType = dict[str, list[object]]
-StatesListsType = list[dict[str, object]]
-ConfigsType = list[tuple[list[callable], list[callable]]]
-EnvProcessesType = dict[str, callable]
+VarDictType = Dict[str, List[object]]
+StatesListsType = List[dict[str, object]]
+ConfigsType = List[tuple[list[callable], List[callable]]]
+EnvProcessesType = Dict[str, callable]
 
 
 def single_proc_exec(

cadCAD/engine/simulation.py

@@ -27,7 +27,7 @@ def policy_scope_tuner(args: tuple,
 
 
 def compose(init_reduction_funct: Aggregator,
-            funct_list: list[Aggregator],
+            funct_list: List[Aggregator],
             val_list: dict) -> object:
     """
     Reduce the nested policy input dict into a simple one.
@@ -57,13 +57,13 @@ def __init__(
 
     def get_policy_input(
         self,
-        sweep_dict: dict[str, list[object]],
+        sweep_dict: Dict[str, List[object]],
         sub_step: int,
-        sL: list[dict[str, object]],
-        s: dict[str, object],
-        funcs: list[callable],
+        sL: List[dict[str, object]],
+        s: Dict[str, object],
+        funcs: List[callable],
         additional_objs
-    ) -> dict[str, object]:
+    ) -> Dict[str, object]:
         """
         Retrieves the Policy Input for usage on State Update Functions
 
@@ -81,7 +81,7 @@ def get_policy_input(
         def execute_policy(f: PolicyFunction) -> dict:
             return policy_scope_tuner(args, additional_objs, f)
 
-        col_results: list[PolicyOutput] = map(execute_policy, funcs)
+        col_results: List[PolicyOutput] = map(execute_policy, funcs)
         # Create a nested dict containing all results combination
         # new_dict[policy_input][policy_ordinal] = policy_input_value
         new_dict: dict = {}
@@ -114,9 +114,9 @@ def f(val_list):
     def apply_env_proc(
         self,
         sweep_dict,
-        env_processes: dict[str, callable],
-        state_dict: dict[str, object]
-    ) -> dict[str, object]:
+        env_processes: Dict[str, callable],
+        state_dict: Dict[str, object]
+    ) -> Dict[str, object]:
 
         def env_composition(target_field, state_dict, target_value):
             function_type = type(lambda x: x)
@@ -149,15 +149,15 @@ def partial_state_update(
         self,
         sweep_dict: Parameters,
         sub_step: Substep,
-        sL: list[State],
+        sL: List[State],
         sH: StateHistory,
         state_funcs: List[StateUpdateFunction],
         policy_funcs: List[PolicyFunction],
         env_processes: EnvProcesses,
         time_step: int,
         run: int,
         additional_objs
-    ) -> list[dict[str, object]]:
+    ) -> List[dict[str, object]]:
 
         if type(additional_objs) == dict:
             if additional_objs.get('deepcopy_off', False) == True:
@@ -210,29 +210,29 @@ def transfer_missing_fields(source, destination):
     # mech_pipeline - state_update_block
     def state_update_pipeline(
         self,
-        sweep_dict: dict[str, list[object]],
+        sweep_dict: Dict[str, List[object]],
         simulation_list,
-        configs: list[tuple[list[callable], list[callable]]],
-        env_processes: dict[str, callable],
+        configs: List[tuple[list[callable], List[callable]]],
+        env_processes: Dict[str, callable],
         time_step: int,
         run: int,
         additional_objs
-    ) -> list[dict[str, object]]:
+    ) -> List[dict[str, object]]:
 
         sub_step = 0
-        states_list_copy: list[dict[str, object]] = tuple(simulation_list[-1])
-        genesis_states: dict[str, object] = states_list_copy[-1].copy()
-#         genesis_states: dict[str, object] = states_list_copy[-1]
+        states_list_copy: List[dict[str, object]] = tuple(simulation_list[-1])
+        genesis_states: Dict[str, object] = states_list_copy[-1].copy()
+#         genesis_states: Dict[str, object] = states_list_copy[-1]
 
         if len(states_list_copy) == 1:
             genesis_states['substep'] = sub_step
 
         del states_list_copy
-        states_list: list[dict[str, object]] = [genesis_states]
+        states_list: List[dict[str, object]] = [genesis_states]
 
         sub_step += 1
         for [s_conf, p_conf] in configs:
-            states_list: list[dict[str, object]] = self.partial_state_update(
+            states_list: List[dict[str, object]] = self.partial_state_update(
                 sweep_dict, sub_step, states_list, simulation_list, s_conf, p_conf, env_processes, time_step, run,
                 additional_objs
             )
@@ -245,19 +245,19 @@ def state_update_pipeline(
     # state_update_pipeline
     def run_pipeline(
         self,
-        sweep_dict: dict[str, list[object]],
-        states_list: list[dict[str, object]],
-        configs: list[tuple[list[callable], list[callable]]],
-        env_processes: dict[str, callable],
+        sweep_dict: Dict[str, List[object]],
+        states_list: List[dict[str, object]],
+        configs: List[tuple[list[callable], List[callable]]],
+        env_processes: Dict[str, callable],
         time_seq: range,
         run: int,
         additional_objs
-    ) -> list[list[dict[str, object]]]:
-        time_seq: list[int] = [x + 1 for x in time_seq]
-        simulation_list: list[list[dict[str, object]]] = [states_list]
+    ) -> List[list[dict[str, object]]]:
+        time_seq: List[int] = [x + 1 for x in time_seq]
+        simulation_list: List[list[dict[str, object]]] = [states_list]
 
         for time_step in time_seq:
-            pipe_run: list[dict[str, object]] = self.state_update_pipeline(
+            pipe_run: List[dict[str, object]] = self.state_update_pipeline(
                 sweep_dict, simulation_list, configs, env_processes, time_step, run, additional_objs
             )
             _, *pipe_run = pipe_run
@@ -283,20 +283,20 @@ def simulation(
         run += 1
         subset_window.appendleft(subset_id)
 
-        def execute_run(sweep_dict, states_list, configs, env_processes, time_seq, _run) -> list[dict[str, object]]:
+        def execute_run(sweep_dict, states_list, configs, env_processes, time_seq, _run) -> List[dict[str, object]]:
             def generate_init_sys_metrics(genesis_states_list, sim_id, _subset_id, _run, _subset_window):
                 for D in genesis_states_list:
                     d = D.copy()
                     d['simulation'], d['subset'], d['run'], d['substep'], d['timestep'] = \
                         sim_id, _subset_id, _run, 0, 0
                     yield d
 
-            states_list_copy: list[dict[str, object]] = list(
+            states_list_copy: List[dict[str, object]] = list(
                 generate_init_sys_metrics(
                     tuple(states_list), simulation_id, subset_id, run, subset_window)
             )
 
-            first_timestep_per_run: list[dict[str, object]] = self.run_pipeline(
+            first_timestep_per_run: List[dict[str, object]] = self.run_pipeline(
                 sweep_dict, states_list_copy, configs, env_processes, time_seq, run, additional_objs
             )
             del states_list_copy

cadCAD/types.py

@@ -3,7 +3,7 @@
 
 State = Dict[str, object]
 Parameters = Dict[str, object]
-SweepableParameters = Dict[str, list[object]]
+SweepableParameters = Dict[str, List[object]]
 Substep = int
 StateHistory = List[List[State]]
 PolicyOutput = Dict[str, object]
@@ -26,7 +26,7 @@ class ConfigurationDict(TypedDict):
 
 TargetValue = object
 EnvProcess: Callable[[State, SweepableParameters, TargetValue], TargetValue]
-EnvProcesses = dict[str, Callable]
+EnvProcesses = Dict[str, Callable]
 TimeSeq = Iterator
 SimulationID = int
 Run = int

cadCAD/utils/__init__.py

@@ -47,7 +47,7 @@ def __call__(self):
         return self.i
 
 
-def compose(*functions: tuple[callable]) -> object:
+def compose(*functions: Tuple[callable]) -> object:
     return reduce(lambda f, g: lambda x: f(g(x)), functions, lambda x: x)
 
 
@@ -100,11 +100,11 @@ def dict_filter(dictionary, condition):
     return dict([(k, v) for k, v in dictionary.items() if condition(v)])
 
 
-def get_max_dict_val_len(g: dict[str, list[int]]) -> int:
+def get_max_dict_val_len(g: Dict[str, List[int]]) -> int:
     return len(max(g.values(), key=len))
 
 
-def tabulate_dict(d: dict[str, list[int]]) -> dict[str, list[int]]:
+def tabulate_dict(d: Dict[str, List[int]]) -> Dict[str, List[int]]:
     max_len = get_max_dict_val_len(d)
     _d = {}
     for k, vl in d.items():
@@ -116,9 +116,9 @@ def tabulate_dict(d: dict[str, list[int]]) -> dict[str, list[int]]:
     return _d
 
 
-def flatten_tabulated_dict(d: dict[str, list[int]]) -> list[dict[str, int]]:
+def flatten_tabulated_dict(d: Dict[str, List[int]]) -> List[dict[str, int]]:
     max_len = get_max_dict_val_len(d)
-    dl: list[dict] = [{} for i in range(max_len)]
+    dl: List[dict] = [{} for i in range(max_len)]
 
     for k, vl in d.items():
         for v, i in zip(vl, list(range(len(vl)))):

documentation/README.md

@@ -31,8 +31,8 @@ exp.append_model(
 Parameters: `append_model`
 * **model_id** : str - OPTIONAL: System Model label
 * **initial_state** : _dict_ - [State Variables](#State-Variables) and their initial values
-* **partial_state_update_blocks** : list[dict[dict]] - list of [Partial State Update Blocks](#Partial-State-Update-Blocks)
-* **policy_ops** : list[functions] - See [Policy Aggregation](Policy_Aggregation.md) 
+* **partial_state_update_blocks** : List[dict[dict]] - list of [Partial State Update Blocks](#Partial-State-Update-Blocks)
+* **policy_ops** : List[functions] - See [Policy Aggregation](Policy_Aggregation.md) 
 * **sim_configs** - See [System Model Parameter Sweep](System_Model_Parameter_Sweep.md)
 * **user_id** : str - OPTIONAL: Configuration User ID
 

documentation/Simulation_Execution.md

@@ -37,7 +37,7 @@ import pandas as pd
 raw_system_events, tensor_field, sessions = simulation.execute()
 
 # Simulation Result Types:
-# raw_system_events: list[dict] 
+# raw_system_events: List[dict] 
 # tensor_field: pd.DataFrame
 
 # Result System Events DataFrame

documentation/examples/param_sweep.py

@@ -14,7 +14,7 @@ def some_function(x):
     return x
 
 
-g: dict[str, list[int]] = {
+g: Dict[str, List[int]] = {
     'alpha': [1],
     'beta': [2, 5],
     'gamma': [3, 4],

testing/models/param_sweep.py

@@ -11,7 +11,7 @@ def some_function(x):
 
 # Optional
 # dict must contain lists opf 2 distinct lengths
-g: dict[str, list[int]] = {
+g: Dict[str, List[int]] = {
     'alpha': [1],
     'beta': [2, some_function],
     'gamma': [3, 4],

testing/tests/test_param_sweep.py

@@ -0,0 +1,21 @@
+from testing.models import param_sweep
+from cadCAD.engine import ExecutionMode, ExecutionContext, Executor
+import pandas as pd # type: ignore
+from testing.results_comparison import dataframe_difference, compare_results_pytest
+import pytest
+
+
+@pytest.fixture
+def ParamSweep():
+    exec_mode = ExecutionMode()
+    exec_ctx = ExecutionContext(context=exec_mode.local_mode)
+    run = Executor(exec_context=exec_ctx, configs=param_sweep.exp.configs)
+    raw_result, _, _ = run.execute()
+
+    result_df = pd.DataFrame(raw_result).
+    expected_df = pd.read_pickle("expected_results/param_sweep_4.pkl")
+    result_diff = dataframe_difference(result_df, expected_df)
+    return result_diff
+
+def test_pytest_compare_results(ParamSweep):
+    compare_results_pytest(ParamSweep)