Skills › BMAD Method › bmad › config_utils.py
config_utils.py
116 lines · skills/bmad/scripts/config_utils.py at bd70c64
1"""Shared strict TOML loading and structural merge support."""
3from __future__ import annotations
5import tomllib
6from collections.abc import Iterable
7from pathlib import Path
8from typing import Any
11class ConfigError(ValueError):
12 """Raised when a present configuration layer cannot be used safely."""
15_KEYED_MERGE_FIELDS = ("code", "id")
18def load_toml(path: Path, *, required: bool = False) -> dict[str, Any]:
19 """Load a TOML table, allowing absence only for optional layers."""
20 if not path.exists():
21 if required:
22 raise ConfigError(f"required TOML file not found: {path}")
23 return {}
24 if not path.is_file():
25 raise ConfigError(f"TOML layer is not a file: {path}")
26 try:
27 with path.open("rb") as stream:
28 parsed = tomllib.load(stream)
29 except tomllib.TOMLDecodeError as error:
30 raise ConfigError(f"failed to parse {path}: {error}") from error
31 except OSError as error:
32 raise ConfigError(f"failed to read {path}: {error}") from error
33 if not isinstance(parsed, dict):
34 raise ConfigError(f"TOML layer did not parse to a table: {path}")
35 return parsed
38def _detect_keyed_merge_field(items: list[Any]) -> str | None:
39 if not items or not all(isinstance(item, dict) for item in items):
40 return None
41 for candidate in _KEYED_MERGE_FIELDS:
42 if all(candidate in item for item in items):
43 for item in items:
44 value = item[candidate]
45 if not isinstance(value, str):
46 raise ConfigError(
47 f"keyed array identifier `{candidate}` must be a string, got {type(value).__name__}"
48 )
49 if not value:
50 raise ConfigError(f"keyed array identifier `{candidate}` must not be empty")
51 return candidate
52 return None
55def _merge_arrays(base: list[Any], override: list[Any]) -> list[Any]:
56 keyed_field = _detect_keyed_merge_field(base + override)
57 if keyed_field is None:
58 return list(base) + list(override)
60 result: list[Any] = []
61 index_by_key: dict[str, int] = {}
62 for item in base:
63 copied = dict(item)
64 index_by_key[copied[keyed_field]] = len(result)
65 result.append(copied)
66 for item in override:
67 copied = dict(item)
68 key = copied[keyed_field]
69 if key in index_by_key:
70 result[index_by_key[key]] = copied
71 else:
72 index_by_key[key] = len(result)
73 result.append(copied)
74 return result
77def structural_merge(base: Any, override: Any) -> Any:
78 """Merge tables recursively, keyed table arrays by identity, and append other arrays."""
79 if isinstance(base, dict) and isinstance(override, dict):
80 result = dict(base)
81 for key, value in override.items():
82 result[key] = structural_merge(result[key], value) if key in result else value
83 return result
84 if isinstance(base, list) and isinstance(override, list):
85 return _merge_arrays(base, override)
86 return override
89def merge_layers(layers: Iterable[dict[str, Any]]) -> dict[str, Any]:
90 merged: dict[str, Any] = {}
91 for layer in layers:
92 merged = structural_merge(merged, layer)
93 return merged
96def load_central_config(project_root: Path) -> dict[str, Any]:
97 bmad_dir = project_root / "_bmad"
98 return merge_layers(
99 (
100 load_toml(bmad_dir / "config.toml", required=True),
101 load_toml(bmad_dir / "custom" / "config.toml"),
102 load_toml(bmad_dir / "custom" / "config.user.toml"),
103 )
104 )
107def load_customization(project_root: Path | None, skill_dir: Path) -> dict[str, Any]:
108 skill_name = skill_dir.name
109 custom_dir = project_root / "_bmad" / "custom" if project_root else None
110 return merge_layers(
111 (
112 load_toml(skill_dir / "customize.toml", required=True),
113 load_toml(custom_dir / f"{skill_name}.toml") if custom_dir else {},
114 load_toml(custom_dir / f"{skill_name}.user.toml") if custom_dir else {},
115 )
116 )