Basic generation and execution progress

Author: CodeByDrescher

poetry.lock

@@ -11,8 +11,9 @@ requires-python = ">=3.10, <4.0"
 dependencies = [
     "process-bigraph (>=0.0.35, <1.0)",
     "typer (>=0.15.3, <1.0)",
-    "process-bigraph-lang @ git+https://github.com/biosimulations/process-bigraph-lang.git",
-    "spatio-flux @ git+https://github.com/vivarium-collective/spatio-flux.git"
+    "process-bigraph-lang @ git+https://github.com/biosimulations/process-bigraph-lang.git#bind-to-object",
+    "spatio-flux @ git+https://github.com/vivarium-collective/spatio-flux.git",
+    "vivarium-interface (>=0.0.5,<0.0.6)"
 ]
 
 [tool.poetry]

pyproject.toml

@@ -1 +1 @@
-from registry import standard_processes
+from registry import spatio_flux_library

src/simple_process_bigraph_runtime/__init__.py

@@ -1,5 +1,6 @@
 import os
 import rich
+from process_bigraph_lang.antlr_dsl.generate import bind_model
 from process_bigraph_lang.dsl import generate
 from process_bigraph_lang.dsl.model import (
     Model,
@@ -9,6 +10,7 @@
 def generateModelAst(pblang_file: os.PathLike[str]) -> Model:
     model_json: str = generate.generate_model(pblang_file)
     model: Model = Model.model_validate_json(model_json)
-    result: str = model.model_dump_json(indent=4)
-    rich.print(result)
+    bind_model(model)
+    #result: str = model.model_dump_json(indent=4)
+    #rich.print(result)
     return model

src/simple_process_bigraph_runtime/dsl_adapter.py

@@ -0,0 +1,83 @@
+import warnings
+import re
+from typing import Any
+
+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] = {}
+        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})`")
+            # create config
+            config = {}
+            # for param in process.config:
+            #     pass
+            # 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}
+
+            # 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)
+            assembler.add_process(
+                name=process.name,
+                process_id=".".join(process_def_actual.python_path.path),
+                #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])
+    assembler.add_emitter()
+
+def determine_builtin_default(type_to_infer: str):
+    if type_to_infer == "float":
+        return 0.0
+    elif type_to_infer == "int":
+        return 0
+    else:
+        raise ValueError(f"Unknown built-in default for type `{type_to_infer}`")

src/simple_process_bigraph_runtime/generation/composite_generator.py

@@ -1,10 +1,17 @@
 import warnings
+from vivarium import Vivarium
+from process_bigraph_lang.dsl.model import Model
 with warnings.catch_warnings(): # Temporary minimal_impact bug in process_bigraphs
     warnings.simplefilter("ignore", SyntaxWarning)
     #warnings.filterwarnings("error", category=SyntaxWarning)
     from process_bigraph import ProcessTypes, Process
 
 
-def generateProcessFromDescription(instructions) -> Process:
-    pass
+def register_process_defs(assembler: Vivarium, model: Model) -> None:
+    # note: does not actually dynamically build processes...yet
+    registry = assembler.core
+    for process_def in model.processDefs:
+        if registry.process_registry.find(process_def.name) is not None:
+            continue
+        raise ValueError(f"Unknown process definition {process_def.name}")
 

src/simple_process_bigraph_runtime/generation/process_generator.py

@@ -1,15 +1,16 @@
 import warnings
-
-from process_bigraph_lang.dsl.model import Model
+from typing import Any
+from process_bigraph_lang.dsl.model import Model, Store
+from vivarium import Vivarium
 
 with warnings.catch_warnings(): # Temporary minimal_impact bug in process_bigraphs
     warnings.simplefilter("ignore", SyntaxWarning)
     #warnings.filterwarnings("error", category=SyntaxWarning)
     from process_bigraph import ProcessTypes
 
 
-def collect_stores(model: Model):
+def collect_stores(model: Model) -> tuple[list[Store], Any]:
     pass
 
-def register_stores(core: ProcessTypes) -> None:
+def register_stores(assembler: Vivarium, stores_to_register: list[Store]) -> None:
     pass

src/simple_process_bigraph_runtime/generation/store_generator.py

@@ -1,19 +1,21 @@
 import warnings
 
 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)
     from process_bigraph import ProcessTypes
 
-
 def collect_types(model: Model) -> list[Type]:
     types_to_process: list[Type] = []
-    for type in model.types:
+    for type_to_add in model.types:
         # Insert some parsing code here
-        types_to_process.append(type)
+        types_to_process.append(type_to_add)
     return types_to_process
 
-def register_type(core: ProcessTypes, types_to_register: list[Type]):
-    pass
+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

src/simple_process_bigraph_runtime/generation/type_generator.py

@@ -1,15 +1,19 @@
 import warnings
 
-from process_bigraph_lang.dsl.model import Model
-
+from process_bigraph_lang.dsl.model import Model, Unit
+from vivarium import Vivarium
 with warnings.catch_warnings(): # Temporary minimal_impact bug in process_bigraphs
     warnings.simplefilter("ignore", SyntaxWarning)
     #warnings.filterwarnings("error", category=SyntaxWarning)
     from process_bigraph import ProcessTypes
 
 
-def collect_units(model: Model):
-    pass
+def collect_units(model: Model) -> list[Unit]:
+    units_to_process: list[Unit] = []
+    for unit in model.units:
+        # Insert some parsing code here
+        units_to_process.append(unit)
+    return units_to_process
 
-def register_unit(core: ProcessTypes, ) -> None:
+def register_units(assembler: Vivarium, units_to_register: list[Unit]) -> None:
     pass

src/simple_process_bigraph_runtime/generation/unit_generator.py

@@ -10,23 +10,56 @@
     #warnings.filterwarnings("error", category=SyntaxWarning)
     from process_bigraph import ProcessTypes
 from simple_process_bigraph_runtime import dsl_adapter
-from spatio_flux import processes
+from vivarium import Vivarium
+from process_bigraph import Composite
 
-app = typer.Typer()
+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
+from bigraph_viz import plot_bigraph
 
+plot_settings = {
+    'remove_process_place_edges': True
+}
+save_images = True
+if save_images:
+    plot_settings.update({
+        'out_dir': '/Users/logandrescher/Documents/pb_output_dir',
+        'dpi': '70'
+    })
 
-core = ProcessTypes()
+app = typer.Typer()
 
 @app.command()
 def validate(pblang_path: Annotated[str, typer.Argument(help="Path to the pblang file to validate")]):
-    generatePythonModel(pblang_path)
+    ast_model: Model = generatePythonModel(pblang_path)
+    _, pb_doc, pb_core = performConversion(ast_model)
+    plot_bigraph(state=pb_doc['state'], schema=pb_doc['composition'],
+                 core=pb_core,
+                 show_values=True,
+                 out_dir=os.path.expanduser("~/Documents/pb_output_dir"),
+                 filename='final_diagram',
+                 dpi='140')
+
+@app.command()
+def execute(duration: float, pblang_path: Annotated[str, typer.Argument(help="Path to the pblang file to validate")]):
+    ast_model: Model = generatePythonModel(pblang_path)
+    pb_composite, _0, _1 = performConversion(ast_model)
+
+    pb_composite.run(duration)
 
 @app.command()
-def execute(pblang_path: Annotated[str, typer.Argument(help="Path to the pblang file to validate")]):
-    generatePythonModel(pblang_path)
+def playground():
+    pass
 
 def generatePythonModel(pblang_path: os.PathLike[str]) -> Model:
-    absolute_path: str = os.path.abspath(pblang_path)
+    absolute_path: str = validate_pb_absolute_path(os.path.abspath(pblang_path))
+
+    return dsl_adapter.generateModelAst(pblang_path)
+
+def validate_pb_absolute_path(absolute_path: str) -> str:
     if not os.path.exists(absolute_path):
         sys.stderr.write(f"Error finding pblang file ({absolute_path}):: no file with that name exists.\n")
         exit(1)
@@ -36,11 +69,17 @@ def generatePythonModel(pblang_path: os.PathLike[str]) -> Model:
     if not absolute_path.endswith(".pblang"):
         sys.stderr.write(f"Error finding pblang file ({absolute_path}):: file is labeled with an unexpected extension.\n")
         exit(1)
+    return absolute_path
 
-    return dsl_adapter.generateModelAst(pblang_path)
+def performConversion(ast_model: Model) -> tuple[Composite, dict, ProcessTypes]:
+    assembler = Vivarium()
+    register_spatio_flux_library(assembler)
+    register_toy_library(assembler)
+    register_types(assembler, collect_types(ast_model))
+    register_units(assembler, collect_units(ast_model))
+    process_composite(ast_model, assembler)
 
-def performConversion(ast_model: Model) -> tuple[str, ProcessTypes]:
-    pass
+    return assembler.composite, assembler.make_document(), assembler.core
 
 if __name__ == "__main__":
     app()