Merge pull request #1 from biosimulators/SimpleRuntime

run toy model from DSL with existing Python Processes

Author: jcschaff

.gitignore

@@ -0,0 +1,3 @@
+*.pyc
+.idea
+.idea/*

poetry.lock

@@ -0,0 +1,2 @@
+[virtualenvs]
+in-project = true

poetry.toml

@@ -0,0 +1,36 @@
+[project]
+name = "simple-process-bigraph-runtime"
+version = "0.1.0"
+description = ""
+authors = [
+    {name = "Logan Drescher",email = "[email protected]"}
+]
+license = {text = "MIT"}
+readme = "README.md"
+requires-python = ">=3.10, <4.0"
+packages = [
+    {include = "simple_process_bigraph_runtime"},
+    {include = "tests"}
+]
+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#bind-to-object",
+    "process-bigraph-lang @ ../process-bigraph-lang",
+    "spatio-flux @ git+https://github.com/vivarium-collective/spatio-flux.git",
+    "vivarium-interface (>=0.0.5,<0.0.6)"
+]
+
+[tool.poetry]
+
+[tool.poetry.group.dev.dependencies]
+pytest = "^7.2.0"
+pytest-cov = "^6.1.1"
+deptry = "^0.23.0"
+mypy = "^1.15.0"
+pre-commit = "^4.2.0"
+tox = "^4.25.0"
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"

pyproject.toml

@@ -0,0 +1 @@
+from simple_process_bigraph_runtime.registry import spatio_flux_library

src/simple_process_bigraph_runtime/__init__.py

@@ -0,0 +1,16 @@
+import os
+
+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,
+)
+
+
+def generateModelAst(pblang_file: os.PathLike[str]) -> Model:
+    model_json: str = generate.generate_model(pblang_file)
+    model: Model = Model.model_validate_json(model_json)
+    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,245 @@
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+from bigraph_schema import is_schema_key, set_path, get_path  # type: ignore[import-untyped]
+from bigraph_schema.utilities import remove_path  # type: ignore[import-untyped]
+from process_bigraph import Composite, Process, Step, ProcessTypes  # type: ignore[import-untyped]
+
+
+class ProcessBigraphEnv:
+    core: ProcessTypes
+    composite: Composite
+    emitter_config: dict[str, Any]
+    emitter_paths: dict[str, tuple[str, ...]]
+
+    def __init__(self):
+        self.core = ProcessTypes()
+        self.composite = Composite(core=self.core)
+        self.emitter_config = {}
+        self.emitter_paths = {}
+
+    def add_process(self, name: str, process_id, config=None, inputs=None, outputs=None,
+                    path: tuple[str, ...] | None = None):
+        edge_type = "process"  # TODO -- does it matter if this is a step or a process?
+        config = config or {}
+        inputs = inputs or {}
+        outputs = outputs or {}
+        path = path or ()
+
+        # convert string paths to lists
+        # TODO -- make this into a separate path-parsing function
+        for ports in [inputs, outputs]:
+            for port, port_path in ports.items():
+                ports[port] = parse_path(port_path)
+
+        # make the process spec
+        state = {
+            name: {
+                "_type": edge_type,
+                "address": f"local:{process_id}",  # TODO -- only support local right now?
+                "config": config,
+                "inputs": inputs,
+                "outputs": outputs,
+            }
+        }
+
+        # nest the process in the composite at the given path
+        self.composite.merge({}, state, path)
+        self.reset_emitters()
+        self.reset_paths()
+
+    def add_step(self, name: str, process_id, config=None, inputs=None, outputs=None,
+                    path: tuple[str, ...] | None = None):
+        edge_type = "step"  # TODO -- does it matter if this is a step or a process?
+        config = config or {}
+        inputs = inputs or {}
+        outputs = outputs or {}
+        path = path or ()
+
+        # convert string paths to lists
+        # TODO -- make this into a separate path-parsing function
+        for ports in [inputs, outputs]:
+            for port, port_path in ports.items():
+                ports[port] = parse_path(port_path)
+
+        # make the process spec
+        state = {
+            name: {
+                "_type": edge_type,
+                "address": f"local:{process_id}",  # TODO -- only support local right now?
+                "config": config,
+                "inputs": inputs,
+                "outputs": outputs,
+            }
+        }
+
+        # nest the process in the composite at the given path
+        self.composite.merge({}, state, path)
+        self.reset_emitters()
+        self.reset_paths()
+
+    def reset_emitters(self) -> None:
+        for path, emitter in self.emitter_paths.items():
+            remove_path(self.composite.state, path)
+            self.add_emitter()
+#         for path, instance in self.composite.step_paths.items():
+#             if "emitter" in path:
+#                 remove_path(self.composite.state, path)
+#                 self.add_emitter()
+
+    def reset_paths(self):
+        self.composite.find_instance_paths(self.composite.state)
+        self.composite.build_step_network()
+
+    def add_emitter(self):
+        """
+        Add an emitter to the composite.
+        """
+
+        emitter_state = self._read_emitter_config()
+
+        # set the emitter at the path
+        path = tuple(self.emitter_config.get("path", ('emitter',)))
+        emitter_state = set_path({}, path, emitter_state)
+
+        self.composite.merge({}, emitter_state)
+
+        # TODO -- this is a hack to get the emitter to show up in the state
+        # TODO -- this should be done in the merge function
+        _, instance = self.core.slice(
+            self.composite.composition,
+            self.composite.state,
+            path)
+
+        self.emitter_paths[path] = instance
+        self.composite.step_paths[path] = instance
+
+        # rebuild the step network
+        self.composite.build_step_network()
+
+    def _read_emitter_config(self):
+        address = self.emitter_config.get("address", "local:ram-emitter")
+        config = self.emitter_config.get("config", {})
+        mode = self.emitter_config.get("mode", "all")
+
+        if mode == "all":
+            inputs = {}
+            for key in self.composite.state.keys():
+                if is_schema_key(key):  # skip schema keys
+                    continue
+                if self.core.inherits_from(self.composite.composition[key], "edge"):  # skip edges
+                    continue
+                inputs[key] = [self.emitter_config.get("inputs", {}).get(key, key)]
+
+        elif mode == "none":
+            inputs = self.emitter_config.get("emit", {})
+
+        elif mode == "bridge":
+            print("Warning: emitter bridge mode not implemented.")
+            inputs = {}
+
+        elif mode == "ports":
+            print("Warning: emitter ports mode not implemented.")
+            inputs = {}
+
+        if not "emit" in config:
+            config["emit"] = {
+                input: "any"
+                for input in inputs}
+
+        return {
+            "_type": "step",
+            "address": address,
+            "config": config,
+            "inputs": inputs}
+
+    def set_value(self, path: list[str], value: Any) -> None:
+        # TODO -- make this set the value in the composite using core
+        set_path(self.composite.state, path=path, value=value)
+        # self.composite[path] = value
+        # self.composite.set({}, value, path)
+
+    def get_type(self, type_id: str) -> dict[str, Any]:
+        return render_type(type_name_or_dict=self.core.access(type_id), core=self.core)
+
+    def make_document(self) -> dict[str, Any]:
+        serialized_state = self.composite.serialize_state()
+
+        # TODO fix RecursionError
+        # schema = self.core.representation(self.composite.composition)
+        schema = self.composite.serialize_schema()
+        # schema = self.composite.composition
+
+        return {
+            "state": serialized_state,
+            "composition": schema,
+        }
+
+    def save(self, json_file_path: Path):
+        document = self.make_document()
+        with open(json_file_path, "w") as json_file:
+            json.dump(document, json_file, indent=4)
+            print(f"Saved file: {json_file_path}")
+
+
+    def run(self, interval):
+        """
+        Run the simulation for a given interval.
+        """
+        if not self.emitter_paths:
+            self.add_emitter()
+
+        self.composite.run(interval)
+
+    def step(self):
+        """
+        Run the simulation for a single step.
+        """
+        self.composite.update({}, 0)
+
+
+
+def render_type(type_name_or_dict: str | dict[str, Any], core: ProcessTypes) -> dict[str, Any]:
+    type_dict: dict[str, Any] = core.access(type_name_or_dict)
+    rendered = {}
+    for key, value in type_dict.items():
+        if is_schema_key(key):
+            if key == '_description':
+                rendered['description'] = value
+            elif key == '_default':
+                rendered['default'] = core.default(type_dict)
+        elif isinstance(value, dict):
+            rendered[key] = render_type(value, core)
+    return rendered
+
+def parse_path(path: list[str] | str | dict[str, Any]) -> list[str] | dict[str, Any]:
+    if isinstance(path, dict):
+        return {
+            key: parse_path(value)
+            for key, value in path.items()}
+    elif isinstance(path, str):
+        if path.startswith('/'):
+            return path.split('/')[1:]
+        else:
+            return [path]
+    else:
+        return path
+
+
+class TypedStep(Step):
+    def __init__(self, config=None):
+        super().__init__(config)
+        self._inputs = self.inputs()
+        self._outputs = self.outputs()
+
+    def inputs(self) -> dict[str, Any]:
+        return {}
+
+
+    def outputs(self) -> dict[str, Any]:
+        return {}
+
+    def update(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        raise NotImplementedError("Subclasses should implement this method.")