use ref_object for object binding

Author: jcschaff

src/simple_process_bigraph_runtime/generation/composite_generator.py

@@ -1,80 +1,54 @@
-import warnings
-import re
-from typing import Any
+from typing import Any, cast
 
+from process_bigraph_lang.dsl.model import Model, StoreDef, ProcessDef, Reference, Store
 from vivarium import Vivarium
-from process_bigraph_lang.dsl.model import Model, StoreDef, ProcessDef, Reference
+
 
 def process_composite(model: Model, assembler: Vivarium):
     for composite_def in model.compositeDefs:
-        store_to_store_def_map: dict[str, StoreDef] = {}
-        for store_ref in composite_def.stores:
-            # verify we have a reference for desired process
-            match_obj = re.match("^#(/(\w+)(@(\d+))+)+$", store_ref.store_def.ref)
-            if not match_obj:
-                raise ValueError(
-                    f"Unexpected reference for process reference: `{store_ref.store_def.ref_text}` ({store_ref.store_def.ref})")
-            full, section, _, index = match_obj.groups()
-            if section != "store_defs":
-                raise ValueError(f"Unexpected section `{section}` for reference `{store_ref.store_def.ref_text}` ({full})`")
-            store_to_store_def_map[composite_def.name + "::" + store_ref.name] = model.store_defs[int(index)]
 
-        store_value_mapping: dict[str, Any] = {}
+        store_path_to_value_map: dict[str, Any] = {}
         for process in composite_def.processes:
-            # verify we have a reference for desired process
-            match_obj = re.match("^#(/(\w+)(@(\d+))+)+$", process.process_def.ref)
-            if not match_obj:
-                raise ValueError(f"Unexpected reference for process reference: `{process.process_def.ref_text}` ({process.process_def.ref})")
-            full, section, _, index = match_obj.groups()
-            if section != "processDefs":
-                raise ValueError(f"Unexpected section `{section}` for reference `{process.process_def.ref_text}` ({full})`")
+            process_def = cast(ProcessDef, process.process_def.ref_object)
+            assert isinstance(process_def, ProcessDef)
 
-            # Identify ports needing to be bound
-            process_def_actual: ProcessDef = model.processDefs[int(index)]
-            input_bindings: dict[str, list[str]] = {elem.ref_text : [] for elem in process_def_actual.inputs}
-            output_bindings: dict[str, list[str]] = {elem.ref_text : [] for elem in process_def_actual.outputs}
+            input_bindings: dict[str, list[str]] = {elem.ref_text : [] for elem in process_def.inputs}
+            output_bindings: dict[str, list[str]] = {elem.ref_text : [] for elem in process_def.outputs}
 
-            # create config
-            # config = {}
-            # for param in process_def_actual.params:
-            #     config[param.name] = param.default.val if param.default is not None else determine_builtin_default(param.type.ref_text)
+            process_config = {}
+            for param in process_def.params:
+                process_config[param.name] = param.default.val if param.default is not None else determine_builtin_default(param.type.ref_text)
 
-            # store validation
-            store_ref: Reference
             for store_ref in process.stores:
-                match_obj = re.match("^#(/[\w]+(@\d)+)+$", store_ref.ref)
-                if not match_obj:
-                    raise ValueError(f"Unexpected reference for process reference: `{store_ref.ref_text}` ({store_ref.ref})")
-                tokenized_reference_path: list[tuple[str, int]] = [(elem.split("@")[0], int(elem.split("@")[1])) for elem in store_ref.ref[1:].split("/") if '@' in elem]
-                map_key: str = composite_def.name + "::" + store_ref.ref_text
-                if map_key not in store_to_store_def_map:
-                    raise ValueError(f"Store {store_ref} not found in mapping using key `{map_key}`")
-                actual_def = store_to_store_def_map[map_key]
-                for state_def in actual_def.states:
-                    in_path = input_bindings[state_def.name] if state_def.name in input_bindings else []
-                    out_binding = output_bindings[state_def.name] if state_def.name in output_bindings else []
-                    if 0 != len(in_path and 0 != len(out_binding)):
-                        continue
-                    if 0 == len(in_path) and 0 == len(out_binding): #nothing has been initialized with this store
-                        store_value_mapping[map_key] = state_def.default.val if state_def.default is not None else determine_builtin_default(state_def.type.ref_text)
-                    if 0 == len(in_path):
-                        for token in map_key.split("::"):
-                            in_path.append(token)
-                    if 0 == len(out_binding):
-                        for token in map_key.split("::"):
-                            out_binding.append(token)
-            config = {p.name: p.default.val for p in process_def_actual.params}
+                store = cast(Store, store_ref.ref_object)
+                assert isinstance(store, Store)
+                store_def = cast(StoreDef, store.store_def.ref_object)
+                assert isinstance(store_def, StoreDef)
+
+                store_path_str: str = composite_def.name + "::" + store.name
+                store_path = [composite_def.name, store.name]
+
+                for state_def in store_def.states:
+                    store_path_to_value_map[store_path_str] = state_def.default.val if state_def.default is not None else determine_builtin_default(state_def.type.ref_text)
+
+                for state_def in store_def.states:
+                    if state_def.name in input_bindings:
+                        input_bindings[state_def.name] = store_path
+                    if state_def.name in output_bindings:
+                        output_bindings[state_def.name] = store_path
+
             assembler.add_process(
                 name=process.name,
-                process_id=".".join(process_def_actual.python_path.path),
-                config=config,
+                process_id=".".join(process_def.python_path.path),
+                config=process_config,
                 inputs=input_bindings,
                 outputs=output_bindings,
             )
-        # end scope of composite def for loop
-        for path_str in store_value_mapping:
-            path = path_str.split("::")
-            assembler.set_value(path=path, value=store_value_mapping[path_str])
+
+        for store_path_str in store_path_to_value_map:
+            store_path = store_path_str.split("::")
+            assembler.set_value(path=store_path, value=store_path_to_value_map[store_path_str])
+
     assembler.add_emitter()
 
 def determine_builtin_default(type_to_infer: str):

src/simple_process_bigraph_runtime/generation/type_generator.py

@@ -2,6 +2,7 @@
 
 from process_bigraph_lang.dsl.model import Model, Type
 from vivarium import Vivarium
+
 with warnings.catch_warnings(): # Temporary minimal_impact bug in process_bigraphs
     warnings.simplefilter("ignore", SyntaxWarning)
     #warnings.filterwarnings("error", category=SyntaxWarning)
@@ -14,8 +15,9 @@ def collect_types(model: Model) -> list[Type]:
         types_to_process.append(type_to_add)
     return types_to_process
 
+
 def register_types(assembler: Vivarium, types_to_register: list[Type]):
     for type_to_register in types_to_register:
         if 0 == assembler.get_type(type_to_register.name).size:
             raise ValueError(f"Type {type_to_register.name} is an unknown type")
-        # in the future, we'd attempt to generate a new type
+        # in the future, we'd attempt to generate a new type

src/simple_process_bigraph_runtime/main.py

@@ -16,8 +16,8 @@
 from simple_process_bigraph_runtime.generation.type_generator import register_types, collect_types
 from simple_process_bigraph_runtime.generation.unit_generator import register_units, collect_units
 from simple_process_bigraph_runtime.generation.composite_generator import process_composite
-from simple_process_bigraph_runtime.registry.spatio_flux_library import apply_to_vivarium as register_spatio_flux_library
-from simple_process_bigraph_runtime.registry.toy_library import apply_to_vivarium as register_toy_library
+import simple_process_bigraph_runtime.registry.spatio_flux_library as spatioflux
+import simple_process_bigraph_runtime.registry.toy_library as toy
 from bigraph_viz import plot_bigraph
 
 plot_settings = {
@@ -74,8 +74,8 @@ def validate_pb_absolute_path(absolute_path: str) -> str:
 
 def performConversion(ast_model: Model) -> tuple[Composite, dict, ProcessTypes]:
     assembler = Vivarium()
-    register_spatio_flux_library(assembler)
-    register_toy_library(assembler)
+    spatioflux.apply_to_vivarium(assembler)
+    toy.apply_to_vivarium(assembler)
     register_types(assembler, collect_types(ast_model))
     register_units(assembler, collect_units(ast_model))
     process_composite(ast_model, assembler)