Config options (#1543)

Co-authored-by: Sam Bull <Sam.B@snowfalltravel.com>
Co-authored-by: Paul Nechifor <paul@nechifor.net>

Author: 3 people

dimos/constants.py

@@ -12,11 +12,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 from pathlib import Path
 
+try:
+    # Not a dependency, just the best way to get config path if available.
+    from gi.repository import GLib  # type: ignore[import-untyped,import-not-found]
+except ImportError:
+    CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
+    STATE_DIR = Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")) / "dimos"
+else:
+    CONFIG_DIR = Path(GLib.get_user_config_dir())
+    STATE_DIR = Path(GLib.get_user_state_dir()) / "dimos"
+
 DIMOS_PROJECT_ROOT = Path(__file__).parent.parent
 
-DIMOS_LOG_DIR = DIMOS_PROJECT_ROOT / "logs"
+if (DIMOS_PROJECT_ROOT / ".git").exists():
+    # Running from Git repository
+    LOG_DIR = DIMOS_PROJECT_ROOT / "logs"
+else:
+    # Running from an installed package - use XDG_STATE_HOME
+    LOG_DIR = STATE_DIR / "logs"
 
 """
 Constants for shared memory

dimos/core/coordination/blueprints.py

@@ -21,9 +21,12 @@
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Literal, Union, get_args, get_origin, get_type_hints
 
+from pydantic import create_model
+
 if TYPE_CHECKING:
     from dimos.protocol.service.system_configurator.base import SystemConfigurator
 
+from dimos.core.global_config import GlobalConfig
 from dimos.core.module import ModuleBase, is_module_type
 from dimos.core.stream import In, Out
 from dimos.core.transport import PubSubTransport
@@ -77,7 +80,7 @@ class ModuleRef:
 
 
 @dataclass(frozen=True)
-class _BlueprintAtom:
+class BlueprintAtom:
     kwargs: dict[str, Any]
     module: type[ModuleBase]
     streams: tuple[StreamRef, ...]
@@ -140,7 +143,7 @@ def create(cls, module: type[ModuleBase], kwargs: dict[str, Any]) -> Self:
 
 @dataclass(frozen=True)
 class Blueprint:
-    blueprints: tuple[_BlueprintAtom, ...]
+    blueprints: tuple[BlueprintAtom, ...]
     disabled_modules_tuple: tuple[type[ModuleBase], ...] = field(default_factory=tuple)
     transport_map: Mapping[tuple[str, type], PubSubTransport[Any]] = field(
         default_factory=lambda: MappingProxyType({})
@@ -154,12 +157,20 @@ class Blueprint:
 
     @classmethod
     def create(cls, module: type[ModuleBase], **kwargs: Any) -> "Blueprint":
-        blueprint = _BlueprintAtom.create(module, kwargs)
+        blueprint = BlueprintAtom.create(module, kwargs)
         return cls(blueprints=(blueprint,))
 
     def disabled_modules(self, *modules: type[ModuleBase]) -> "Blueprint":
         return replace(self, disabled_modules_tuple=self.disabled_modules_tuple + modules)
 
+    def config(self) -> type:
+        configs = {
+            b.module.name: (get_type_hints(b.module)["config"] | None, None)
+            for b in self.blueprints
+        }
+        configs["g"] = (GlobalConfig | None, None)
+        return create_model("BlueprintConfig", __config__={"extra": "forbid"}, **configs)  # type: ignore[call-overload,no-any-return]
+
     def transports(self, transports: dict[tuple[str, type], Any]) -> "Blueprint":
         return replace(self, transport_map=MappingProxyType({**self.transport_map, **transports}))
 
@@ -185,7 +196,7 @@ def configurators(self, *checks: "SystemConfigurator") -> "Blueprint":
         return replace(self, configurator_checks=self.configurator_checks + tuple(checks))
 
     @cached_property
-    def active_blueprints(self) -> tuple[_BlueprintAtom, ...]:
+    def active_blueprints(self) -> tuple[BlueprintAtom, ...]:
         if not self.disabled_modules_tuple:
             return self.blueprints
         disabled = set(self.disabled_modules_tuple)
@@ -219,7 +230,7 @@ def autoconnect(*blueprints: Blueprint) -> Blueprint:
     )
 
 
-def _eliminate_duplicates(blueprints: list[_BlueprintAtom]) -> list[_BlueprintAtom]:
+def _eliminate_duplicates(blueprints: list[BlueprintAtom]) -> list[BlueprintAtom]:
     # The duplicates are eliminated in reverse so that newer blueprints override older ones.
     seen = set()
     unique_blueprints = []

dimos/core/coordination/module_coordinator.py

@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 from collections import defaultdict
-from collections.abc import Mapping
+from collections.abc import Mapping, MutableMapping
 import importlib
 import shutil
 import sys
@@ -35,7 +35,7 @@
 from dimos.utils.safe_thread_map import safe_thread_map
 
 if TYPE_CHECKING:
-    from dimos.core.coordination.blueprints import Blueprint, _BlueprintAtom
+    from dimos.core.coordination.blueprints import Blueprint, BlueprintAtom
     from dimos.core.rpc_client import ModuleProxy, ModuleProxyProtocol
 
 logger = setup_logger()
@@ -56,7 +56,7 @@ def __init__(
             cls.deployment_identifier: cls(g=g) for cls in manager_types
         }
         self._deployed_modules = {}
-        self._deployed_atoms: dict[type[ModuleBase], _BlueprintAtom] = {}
+        self._deployed_atoms: dict[type[ModuleBase], BlueprintAtom] = {}
         self._resolved_module_refs: dict[tuple[type[ModuleBase], str], type[ModuleBase]] = {}
         self._transport_registry: dict[tuple[str, type], PubSubTransport[Any]] = {}
         self._class_aliases: dict[type[ModuleBase], type[ModuleBase]] = {}
@@ -114,7 +114,9 @@ def deploy(
         self._deployed_modules[module_class] = deployed_module
         return deployed_module  # type: ignore[return-value]
 
-    def deploy_parallel(self, module_specs: list[ModuleSpec]) -> list[ModuleProxy]:
+    def deploy_parallel(
+        self, module_specs: list[ModuleSpec], blueprint_args: Mapping[str, Mapping[str, Any]]
+    ) -> list[ModuleProxy]:
         if not self._managers:
             raise ValueError("Not started")
 
@@ -130,7 +132,7 @@ def deploy_parallel(self, module_specs: list[ModuleSpec]) -> list[ModuleProxy]:
         results: list[Any] = [None] * len(module_specs)
 
         def _deploy_group(dep: str) -> None:
-            deployed = self._managers[dep].deploy_parallel(specs_by_deployment[dep])
+            deployed = self._managers[dep].deploy_parallel(specs_by_deployment[dep], blueprint_args)
             for index, module in zip(indices_by_deployment[dep], deployed, strict=True):
                 results[index] = module
 
@@ -221,12 +223,13 @@ def _connect_streams(self, blueprint: Blueprint) -> None:
     def build(
         cls,
         blueprint: Blueprint,
-        cli_config_overrides: Mapping[str, Any] | None = None,
+        blueprint_args: MutableMapping[str, Any] | None = None,
     ) -> ModuleCoordinator:
         logger.info("Building the blueprint")
         global_config.update(**dict(blueprint.global_config_overrides))
-        if cli_config_overrides:
-            global_config.update(**dict(cli_config_overrides))
+        blueprint_args = blueprint_args or {}
+        if "g" in blueprint_args:
+            global_config.update(**blueprint_args.pop("g"))
 
         _run_configurators(blueprint)
         _check_requirements(blueprint)
@@ -236,7 +239,7 @@ def build(
         coordinator = cls(g=global_config)
         coordinator.start()
 
-        _deploy_all_modules(blueprint, coordinator, global_config)
+        _deploy_all_modules(blueprint, coordinator, global_config, blueprint_args)
         coordinator._connect_streams(blueprint)
         _connect_module_refs(blueprint, coordinator)
 
@@ -250,7 +253,7 @@ def build(
     def load_blueprint(
         self,
         blueprint: Blueprint,
-        cli_config_overrides: Mapping[str, Any] | None = None,
+        blueprint_args: MutableMapping[str, Mapping[str, Any]] | None = None,
     ) -> None:
         """Load a blueprint into an already-running coordinator.
 
@@ -263,8 +266,9 @@ def load_blueprint(
 
         # Apply config overrides.
         self._global_config.update(**dict(blueprint.global_config_overrides))
-        if cli_config_overrides:
-            self._global_config.update(**dict(cli_config_overrides))
+        blueprint_args = blueprint_args or {}
+        if "g" in blueprint_args:
+            self._global_config.update(**blueprint_args.pop("g"))
 
         # Scale worker pool.
         n_extra = int(blueprint.global_config_overrides.get("n_workers", 0))
@@ -288,7 +292,7 @@ def load_blueprint(
 
         before = set(self._deployed_modules)
 
-        _deploy_all_modules(blueprint, self, self._global_config)
+        _deploy_all_modules(blueprint, self, self._global_config, blueprint_args)
         self._connect_streams(blueprint)
         _connect_module_refs(blueprint, self, existing_modules=before)
 
@@ -300,8 +304,12 @@ def load_blueprint(
 
         self._send_on_system_modules()
 
-    def load_module(self, module_class: type[ModuleBase], **kwargs: Any) -> None:
-        self.load_blueprint(module_class.blueprint(**kwargs))
+    def load_module(
+        self,
+        module_class: type[ModuleBase],
+        blueprint_args: MutableMapping[str, Mapping[str, Any]] | None = None,
+    ) -> None:
+        self.load_blueprint(module_class.blueprint(**blueprint_args or {}))
 
     def unload_module(self, module_class: type[ModuleBase]) -> None:
         """Stop and tear down a single deployed module.
@@ -576,13 +584,16 @@ def _check_requirements(blueprint: Blueprint) -> None:
 
 
 def _deploy_all_modules(
-    blueprint: Blueprint, module_coordinator: ModuleCoordinator, gc: GlobalConfig
+    blueprint: Blueprint,
+    module_coordinator: ModuleCoordinator,
+    gc: GlobalConfig,
+    blueprint_args: Mapping[str, Mapping[str, Any]],
 ) -> None:
     module_specs: list[ModuleSpec] = []
     for bp in blueprint.active_blueprints:
-        module_specs.append((bp.module, gc, bp.kwargs))
+        module_specs.append((bp.module, gc, bp.kwargs.copy()))
 
-    module_coordinator.deploy_parallel(module_specs)
+    module_coordinator.deploy_parallel(module_specs, blueprint_args)
 
     for bp in blueprint.active_blueprints:
         module_coordinator._deployed_atoms[bp.module] = bp

dimos/core/coordination/test_blueprints.py

@@ -14,7 +14,10 @@
 
 
 import pickle
-from typing import Protocol
+from typing import Protocol, get_type_hints
+
+from pydantic import ValidationError
+import pytest
 
 from dimos.core._test_future_annotations_helper import (
     FutureData,
@@ -23,10 +26,10 @@
 )
 from dimos.core.coordination.blueprints import (
     Blueprint,
+    BlueprintAtom,
     DisabledModuleProxy,
     ModuleRef,
     StreamRef,
-    _BlueprintAtom,
     autoconnect,
 )
 from dimos.core.core import rpc
@@ -83,7 +86,7 @@ def what_is_as_name(self) -> str:
 
 
 def test_get_connection_set() -> None:
-    assert _BlueprintAtom.create(CatModule, kwargs={"k": "v"}) == _BlueprintAtom(
+    assert BlueprintAtom.create(CatModule, kwargs={"k": "v"}) == BlueprintAtom(
         module=CatModule,
         streams=(
             StreamRef(name="pet_cat", type=Petting, direction="in"),
@@ -99,7 +102,7 @@ def test_autoconnect() -> None:
 
     assert blueprint_set == Blueprint(
         blueprints=(
-            _BlueprintAtom(
+            BlueprintAtom(
                 module=ModuleA,
                 streams=(
                     StreamRef(name="data1", type=Data1, direction="out"),
@@ -108,7 +111,7 @@ def test_autoconnect() -> None:
                 module_refs=(),
                 kwargs={},
             ),
-            _BlueprintAtom(
+            BlueprintAtom(
                 module=ModuleB,
                 streams=(
                     StreamRef(name="data1", type=Data1, direction="in"),
@@ -122,6 +125,17 @@ def test_autoconnect() -> None:
     )
 
 
+def test_config() -> None:
+    blueprint = autoconnect(ModuleA.blueprint(), ModuleB.blueprint())
+    config = blueprint.config()
+    assert config.model_fields.keys() == {"modulea", "moduleb", "g"}
+    assert config.model_fields["modulea"].annotation == get_type_hints(ModuleA)["config"] | None
+    assert config.model_fields["moduleb"].annotation == get_type_hints(ModuleB)["config"] | None
+
+    with pytest.raises(ValidationError, match="invalid_key"):
+        config(module_a={"invalid_key": 5})
+
+
 def test_transports() -> None:
     custom_transport = LCMTransport("/custom_topic", Data1)
     blueprint_set = autoconnect(ModuleA.blueprint(), ModuleB.blueprint()).transports(
@@ -147,16 +161,16 @@ def test_future_annotations_support() -> None:
     """Test that modules using `from __future__ import annotations` work correctly.
 
     PEP 563 (future annotations) stores annotations as strings instead of actual types.
-    This test verifies that _BlueprintAtom.create properly resolves string annotations
+    This test verifies that BlueprintAtom.create properly resolves string annotations
     to the actual In/Out types.
     """
 
     # Test that streams are properly extracted from modules with future annotations
-    out_blueprint = _BlueprintAtom.create(FutureModuleOut, kwargs={})
+    out_blueprint = BlueprintAtom.create(FutureModuleOut, kwargs={})
     assert len(out_blueprint.streams) == 1
     assert out_blueprint.streams[0] == StreamRef(name="data", type=FutureData, direction="out")
 
-    in_blueprint = _BlueprintAtom.create(FutureModuleIn, kwargs={})
+    in_blueprint = BlueprintAtom.create(FutureModuleIn, kwargs={})
     assert len(in_blueprint.streams) == 1
     assert in_blueprint.streams[0] == StreamRef(name="data", type=FutureData, direction="in")
 
@@ -186,7 +200,7 @@ class ModuleWithOptionalRef(Module):
 
 
 def test_optional_module_ref_detected() -> None:
-    atom = _BlueprintAtom.create(ModuleWithOptionalRef, kwargs={})
+    atom = BlueprintAtom.create(ModuleWithOptionalRef, kwargs={})
     assert len(atom.module_refs) == 1
     ref = atom.module_refs[0]
     assert ref.name == "calc"