feat: add collect-env CLI subcommand with formatted summary
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
Executable
+202
@@ -0,0 +1,202 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Rockchip Agents 启动入口"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Rockchip Agents")
|
||||
sub = parser.add_subparsers(dest="cmd")
|
||||
sub.add_parser("scan", help="扫描所有项目，打印发现的任务（不执行）")
|
||||
sub.add_parser("run", help="扫描 → 入队 → 执行所有任务（一键完成）")
|
||||
sub.add_parser("run-next", help="执行队列中下一个已入队任务")
|
||||
sub.add_parser("status", help="查看任务队列状态")
|
||||
sub.add_parser("start", help="启动定时调度器（每小时自动执行）")
|
||||
p_av = sub.add_parser("arch-versions", help="查看项目的架构优化版本历史")
|
||||
p_av.add_argument("project", help="项目名称（如 yolo、mediapipe）")
|
||||
p_ar = sub.add_parser("arch-restore", help="将项目恢复到指定架构版本")
|
||||
p_ar.add_argument("project", help="项目名称")
|
||||
p_ar.add_argument("version_id", help="版本 ID（如 v001）")
|
||||
sub.add_parser("goals", help="查看所有项目的长期目标记忆")
|
||||
sub.add_parser("facts", help="查看所有项目的知识库事实（由 agents 自动记录）")
|
||||
p_sf = sub.add_parser("set-fact", help="手动写入项目事实")
|
||||
p_sf.add_argument("project", help="项目名")
|
||||
p_sf.add_argument("key", help="事实键名，如 test_cmd")
|
||||
p_sf.add_argument("value", help="事实值")
|
||||
sub.add_parser("projects", help="查看自动发现的所有项目（含类型/模式）")
|
||||
sub.add_parser("collect-env", help="收集本机+SSH设备环境信息（模型路径/磁盘/包）")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.cmd == "scan":
|
||||
from rockchip_agents.core.scanner import Scanner
|
||||
from rockchip_agents.config import load_config
|
||||
cfg = load_config()
|
||||
tasks = Scanner(cfg).scan_all()
|
||||
if not tasks:
|
||||
print("未发现待处理任务")
|
||||
for t in tasks:
|
||||
print(f"[{t.project}] {t.title} (priority={t.priority})")
|
||||
|
||||
elif args.cmd == "run":
|
||||
import asyncio
|
||||
from rockchip_agents.agents.manager import ManagerAgent
|
||||
from rockchip_agents.config import load_config
|
||||
cfg = load_config()
|
||||
stats = asyncio.run(ManagerAgent(cfg).run_cycle())
|
||||
print(f"扫描入队: {stats['enqueued']} 个，执行完成: {stats['executed']} 个")
|
||||
|
||||
elif args.cmd == "run-next":
|
||||
import asyncio
|
||||
from rockchip_agents.core.executor import Executor
|
||||
from rockchip_agents.config import load_config
|
||||
cfg = load_config()
|
||||
ran = asyncio.run(Executor(cfg).run_next())
|
||||
print("已执行" if ran else "队列为空")
|
||||
|
||||
elif args.cmd == "status":
|
||||
from rockchip_agents.agents.manager import ManagerAgent
|
||||
from rockchip_agents.config import load_config
|
||||
cfg = load_config()
|
||||
status = ManagerAgent(cfg).get_status()
|
||||
if not status:
|
||||
print("队列为空")
|
||||
else:
|
||||
for state, count in status.items():
|
||||
print(f" {state}: {count}")
|
||||
|
||||
elif args.cmd == "start":
|
||||
from rockchip_agents.core.scheduler import start
|
||||
start()
|
||||
|
||||
elif args.cmd == "arch-versions":
|
||||
from rockchip_agents.config import load_config
|
||||
from rockchip_agents.tools.arch_version import ArchVersion
|
||||
cfg = load_config()
|
||||
proj_cfg = cfg.projects.get(args.project)
|
||||
if not proj_cfg:
|
||||
print(f"项目 {args.project!r} 不在配置中")
|
||||
else:
|
||||
av = ArchVersion(proj_cfg.path)
|
||||
versions = av.list_versions()
|
||||
if not versions:
|
||||
print(f"{args.project}: 暂无架构优化版本")
|
||||
else:
|
||||
best = av.current_best()
|
||||
for v in versions:
|
||||
mark = " ★" if v.version_id == best else ""
|
||||
passed = "✓" if v.test_passed else ("✗" if v.test_passed is False else "?")
|
||||
print(f" {v.version_id}{mark} [{v.status:12s}] [{passed}] {v.created_at[:16]} {v.description}")
|
||||
|
||||
elif args.cmd == "arch-restore":
|
||||
from rockchip_agents.config import load_config
|
||||
from rockchip_agents.tools.arch_version import ArchVersion
|
||||
cfg = load_config()
|
||||
proj_cfg = cfg.projects.get(args.project)
|
||||
if not proj_cfg:
|
||||
print(f"项目 {args.project!r} 不在配置中")
|
||||
else:
|
||||
av = ArchVersion(proj_cfg.path)
|
||||
av.restore(args.version_id)
|
||||
print(f"已将 {args.project} 恢复到 {args.version_id}")
|
||||
|
||||
elif args.cmd == "goals":
|
||||
from rockchip_agents.tools.project_memory import ProjectMemory
|
||||
goals = ProjectMemory().list_goals()
|
||||
if not goals:
|
||||
print("暂无项目目标记录（运行 productizer 或 architect 任务后自动写入）")
|
||||
else:
|
||||
for proj, g in goals.items():
|
||||
print(f"\n[{proj}] by {g.updated_by} {g.updated_at[:10]}")
|
||||
print(f" {g.goal_text}")
|
||||
|
||||
elif args.cmd == "facts":
|
||||
from rockchip_agents.tools.project_memory import ProjectMemory
|
||||
from rockchip_agents.config import load_config
|
||||
mem = ProjectMemory()
|
||||
cfg = load_config()
|
||||
any_found = False
|
||||
for proj_name in sorted(cfg.projects):
|
||||
facts = mem.get_facts(proj_name)
|
||||
if facts:
|
||||
any_found = True
|
||||
print(f"\n[{proj_name}]")
|
||||
for k, v in facts.items():
|
||||
print(f" {k}: {v}")
|
||||
if not any_found:
|
||||
print("暂无项目事实记录（运行 architect 任务后自动写入）")
|
||||
|
||||
elif args.cmd == "set-fact":
|
||||
from rockchip_agents.tools.project_memory import ProjectMemory
|
||||
mem = ProjectMemory()
|
||||
mem.set_fact(args.project, args.key, args.value, role="developer")
|
||||
print(f"已写入: [{args.project}] {args.key} = {args.value}")
|
||||
|
||||
elif args.cmd == "projects":
|
||||
from rockchip_agents.config import load_config
|
||||
cfg = load_config()
|
||||
print(f"{'项目名':20s} {'类型':8s} {'模式':8s} 路径")
|
||||
print("-" * 70)
|
||||
for name, p in sorted(cfg.projects.items()):
|
||||
dw = f" → {p.device_workspace}" if p.device_workspace else ""
|
||||
print(f"{name:20s} {p.kind:8s} {p.mode:8s} {p.path}{dw}")
|
||||
|
||||
elif args.cmd == "collect-env":
|
||||
from rockchip_agents.config import load_config
|
||||
from rockchip_agents.tools.env_collector import run_collection
|
||||
from rockchip_agents.tools.project_memory import ProjectMemory
|
||||
cfg = load_config()
|
||||
mem = ProjectMemory()
|
||||
facts = run_collection(cfg, mem)
|
||||
|
||||
# 本机摘要
|
||||
env_vars = {k: v for k, v in facts.items() if k.startswith("local.envvar.")}
|
||||
print("\n[本机]")
|
||||
if env_vars:
|
||||
print(f" ENV: {len(env_vars)} 条含模型关键词变量")
|
||||
for k, v in list(env_vars.items())[:3]:
|
||||
print(f" {k.split('.')[-1]}={v}")
|
||||
for ext in ("rknn", "onnx", "pt", "pb"):
|
||||
val = facts.get(f"local.models.{ext}", "")
|
||||
if val:
|
||||
cnt = len([p for p in val.split(",") if p])
|
||||
print(f" *.{ext}: {cnt} 个文件")
|
||||
|
||||
# 各设备摘要
|
||||
device_names: set[str] = set()
|
||||
for key in facts:
|
||||
parts = key.split(".")
|
||||
if len(parts) >= 2 and parts[0] not in ("local", "collected"):
|
||||
device_names.add(parts[0])
|
||||
|
||||
for dev_name in sorted(device_names):
|
||||
dev_cfg = cfg.devices.get(dev_name)
|
||||
host_info = f" @ {dev_cfg.host}" if dev_cfg else ""
|
||||
print(f"\n[{dev_name}{host_info}]")
|
||||
error = facts.get(f"{dev_name}.error", "")
|
||||
if error:
|
||||
print(f" ⚠ {error}")
|
||||
continue
|
||||
rknn_val = facts.get(f"{dev_name}.models.rknn", "")
|
||||
if rknn_val:
|
||||
cnt = len([p for p in rknn_val.split(",") if p])
|
||||
print(f" *.rknn: {cnt} 个文件")
|
||||
disk = facts.get(f"{dev_name}.disk", "")
|
||||
if disk:
|
||||
print(f" 磁盘: {disk[:80]}")
|
||||
pkgs = facts.get(f"{dev_name}.packages", "")
|
||||
if pkgs:
|
||||
print(f" 关键包: {pkgs[:80]}")
|
||||
|
||||
ts = facts.get("collected_at", "")
|
||||
print(f"\n已保存到 ProjectMemory (__env__) [{ts}]")
|
||||
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user