aureon-os / scripts /aureon_capability_scanner.py
GaryLeckey's picture
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified
Raw History Blame Contribute Delete
10.7 kB
#!/usr/bin/env python3
"""
AUREON CAPABILITY SCANNER
═════════════════════════
Discovers and validates ALL Aureon Python modules, reporting:
- Import status (can the module be imported?)
- Available classes and functions
- Key entry points and CLI capabilities
- Integration bridges (Ollama, Obsidian, World Data)
- Coding skills and architect primitives
- LLM adapters and agent runners
Usage:
python scripts/aureon_capability_scanner.py
python scripts/aureon_capability_scanner.py --json
python scripts/aureon_capability_scanner.py --focus coding,llm,integrations
"""
from __future__ import annotations
import argparse
import importlib
import inspect
import json
import sys
import traceback
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
ROOT = Path(__file__).resolve().parents[1]
PACKAGE = ROOT / "aureon"
@dataclass
class CapabilityModule:
module_path: str
import_ok: bool = False
error: Optional[str] = None
classes: List[str] = field(default_factory=list)
functions: List[str] = field(default_factory=list)
has_main: bool = False
entry_points: List[str] = field(default_factory=list)
@dataclass
class CapabilityReport:
generated_at: str
root: str
total_modules: int = 0
import_ok: int = 0
import_failed: int = 0
modules: List[CapabilityModule] = field(default_factory=list)
categories: Dict[str, Any] = field(default_factory=dict)
summary: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"generated_at": self.generated_at,
"root": self.root,
"total_modules": self.total_modules,
"import_ok": self.import_ok,
"import_failed": self.import_failed,
"modules": [asdict(m) for m in self.modules],
"categories": self.categories,
"summary": self.summary,
}
# Modules we care most about for the murge integration
FOCUS_MODULES: Dict[str, List[str]] = {
"coding": [
"aureon.code_architect.architect",
"aureon.code_architect.executor",
"aureon.code_architect.observer",
"aureon.code_architect.skill_library",
"aureon.code_architect.validator",
"aureon.code_architect.writer",
"aureon.autonomous.aureon_repo_task_bridge",
"aureon.autonomous.aureon_safe_code_control",
"aureon.core.goal_execution_engine",
],
"llm": [
"aureon.inhouse_ai.llm_adapter",
"aureon.inhouse_ai.agent_runner",
"aureon.inhouse_ai.orchestrator",
"aureon.inhouse_ai.agent",
"aureon.inhouse_ai.agent_pool",
"aureon.inhouse_ai.task_queue",
"aureon.inhouse_ai.team",
"aureon.integrations.ollama.ollama_bridge",
"aureon.integrations.ollama.ollama_adapter",
],
"integrations": [
"aureon.integrations.obsidian.obsidian_bridge",
"aureon.integrations.obsidian.obsidian_sink",
"aureon.integrations.world_data.world_data_ingester",
"aureon.integrations.wiring",
"aureon.integrations.audit_trail",
"aureon.integrations.neural_pathway_mapper",
],
"queen": [
"aureon.queen.queen_action_bridge",
"aureon.queen.queen_prose_composer",
"aureon.queen.self_enhancement_engine",
"aureon.queen.knowledge_interpreter",
"aureon.queen.temporal_knowledge",
"aureon.queen.vault_knowledge_bridge",
"aureon.queen.conversation_memory",
"aureon.queen.meaning_resolver",
],
"vault": [
"aureon.vault.aureon_vault",
"aureon.vault.voice.vault_voice",
"aureon.vault.voice.self_dialogue",
"aureon.vault.auris_metacognition",
"aureon.vault.self_feedback_loop",
"aureon.vault.hnc_deployer",
],
"autonomous": [
"aureon.autonomous.aureon_autonomy_hub",
"aureon.autonomous.aureon_full_orchestrator",
"aureon.autonomous.aureon_queen_autonomous",
"aureon.autonomous.aureon_repo_task_bridge",
"aureon.autonomous.vm_control.dispatcher",
],
}
def discover_modules() -> List[str]:
modules: List[str] = []
if not PACKAGE.exists():
return modules
for path in PACKAGE.rglob("*.py"):
if "__pycache__" in path.parts or path.name.startswith("_"):
continue
rel = path.relative_to(PACKAGE)
parts = list(rel.with_suffix("").parts)
if parts[-1] == "__init__":
parts = parts[:-1]
module_name = "aureon." + ".".join(parts)
modules.append(module_name)
return sorted(set(modules))
def scan_module(module_name: str) -> CapabilityModule:
result = CapabilityModule(module_path=module_name)
try:
mod = importlib.import_module(module_name)
result.import_ok = True
for name, obj in inspect.getmembers(mod):
if inspect.isclass(obj) and obj.__module__ == module_name:
result.classes.append(name)
elif inspect.isfunction(obj) and obj.__module__ == module_name:
result.functions.append(name)
# Check for main block
source = inspect.getsource(mod)
result.has_main = '__name__ == "__main__"' in source or "if __name__" in source
except Exception as exc:
result.import_ok = False
result.error = f"{type(exc).__name__}: {exc}"
return result
def run_scan(args: argparse.Namespace) -> CapabilityReport:
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
report = CapabilityReport(
generated_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
root=str(ROOT),
)
focus_sets: List[str] = []
if args.focus:
focus_sets = [s.strip() for s in args.focus.split(",")]
if focus_sets:
module_names = []
for cat in focus_sets:
module_names.extend(FOCUS_MODULES.get(cat, []))
module_names = sorted(set(module_names))
else:
module_names = discover_modules()
report.total_modules = len(module_names)
for name in module_names:
mod = scan_module(name)
report.modules.append(mod)
if mod.import_ok:
report.import_ok += 1
else:
report.import_failed += 1
# Build category summaries
for cat, mods in FOCUS_MODULES.items():
found = [m for m in report.modules if m.module_path in mods]
report.categories[cat] = {
"total": len(mods),
"import_ok": sum(1 for m in found if m.import_ok),
"import_failed": sum(1 for m in found if not m.import_ok),
"modules": [asdict(m) for m in found],
}
report.summary = {
"pass_rate_percent": round(report.import_ok / max(report.total_modules, 1) * 100, 1),
"coding_ready": report.categories.get("coding", {}).get("import_ok", 0) > 0,
"llm_ready": report.categories.get("llm", {}).get("import_ok", 0) > 0,
"integrations_ready": report.categories.get("integrations", {}).get("import_ok", 0) > 0,
"queen_ready": report.categories.get("queen", {}).get("import_ok", 0) > 0,
"vault_ready": report.categories.get("vault", {}).get("import_ok", 0) > 0,
"autonomous_ready": report.categories.get("autonomous", {}).get("import_ok", 0) > 0,
}
return report
def main() -> int:
ap = argparse.ArgumentParser(description="Scan Aureon capabilities")
ap.add_argument("--json", action="store_true", help="Emit JSON only")
ap.add_argument("--focus", default="", help="Comma-separated categories: coding,llm,integrations,queen,vault,autonomous")
args = ap.parse_args()
report = run_scan(args)
if args.json:
print(json.dumps(report.to_dict(), indent=2))
return 0 if report.import_failed == 0 else 1
print("═══════════════════════════════════════════════════════════════════════════════")
print(" AUREON CAPABILITY SCANNER")
print("═══════════════════════════════════════════════════════════════════════════════")
print(f" Root: {report.root}")
print(f" Modules scanned: {report.total_modules}")
print(f" Import OK: {report.import_ok}")
print(f" Import failed: {report.import_failed}")
print(f" Pass rate: {report.summary['pass_rate_percent']}%")
print("")
for cat, data in report.categories.items():
ok = data["import_ok"]
total = data["total"]
status = "βœ“" if ok == total else "~" if ok > 0 else "βœ—"
print(f" {status} {cat.upper():14} {ok}/{total} modules importable")
for m in data["modules"]:
if not m["import_ok"]:
print(f" βœ— {m['module_path']}: {m['error'][:80]}")
print("")
print("═══════════════════════════════════════════════════════════════════════════════")
print(f" CODING: {'READY' if report.summary['coding_ready'] else 'ATTENTION'}")
print(f" LLM: {'READY' if report.summary['llm_ready'] else 'ATTENTION'}")
print(f" INTEGRATIONS: {'READY' if report.summary['integrations_ready'] else 'ATTENTION'}")
print(f" QUEEN: {'READY' if report.summary['queen_ready'] else 'ATTENTION'}")
print(f" VAULT: {'READY' if report.summary['vault_ready'] else 'ATTENTION'}")
print(f" AUTONOMOUS: {'READY' if report.summary['autonomous_ready'] else 'ATTENTION'}")
print("═══════════════════════════════════════════════════════════════════════════════")
return 0 if report.import_failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())