from __future__ import annotations import time from typing import Any from .acceptance import ( latest_acceptance_report, latest_deep_acceptance_report, latest_real_acceptance_report, latest_real_train_acceptance_report, ) from .catalog import get_catalog from .coverage import get_coverage_report from .modules.dataset.service import list_uploaded_datasets from .modules.results.service import scan_results, scan_training_curves from .modules.system.service import get_gpus, get_runtime_readiness from .modules.weights.service import load_manifest CAPABILITY_GROUPS = [ { "id": "dataset", "label": "Dataset", "description": "上传图片、label、mask，并执行预处理、配对、叠加、拼接、视频抽帧和 YOLO 数据构建。", "required_tasks": [ "dataset.rename", "dataset.to_png", "dataset.resize", "dataset.pair", "dataset.rebuild_labels", "dataset.stack", "dataset.stitch", "dataset.video_frames", "dataset.yolo_txt_sort", "dataset.yolo_convert_png", "dataset.yolo_resize", ], "task_prefixes": ["dataset."], "evidence_roles": ["segmentation"], }, { "id": "segmodel", "label": "SegModel", "description": "SMP/SegModel 单模型和批量训练、预测、raw mask、指标、FLOPs/FPS。", "required_tasks": [ "segmodel.train", "segmodel.batch_train", "segmodel.predict", "segmodel.batch_predict", "segmodel.flops", "segmodel.benchmark", "segmodel.raw_mask_check", "segmodel.metrics", ], "task_prefixes": ["segmodel."], "runtime_roles": ["backend_task"], "families": ["segmodel"], }, { "id": "yolo", "label": "YOLO", "description": "YOLOv8/v9/v11/v12 分割训练、上传数据集训练、预测、热度图、对比、视频预测和 raw mask。", "required_tasks": [ "yolo.train", "yolo.train_custom", "yolo.predict", "yolo.predict_custom", "yolo.batch_predict", "yolo.heatmap", "yolo.heatmap_custom", "yolo.compare", "yolo.raw_mask_check", "yolo.video_visible", "yolo.video_unvisible", ], "task_prefixes": ["yolo."], "runtime_roles": ["backend_task"], "families": ["yolo"], "evidence_roles": ["heatmap", "segmentation", "curve", "video"], }, { "id": "mmseg", "label": "MMSeg", "description": "MMSeg 数据/算法配置生成、31 个算法、训练、预测、指标、FLOPs/FPS、绘图和 loss/mIoU 提取。", "required_tasks": [ "mmseg.generate_data", "mmseg.generate_alg", "mmseg.train", "mmseg.predict_v1", "mmseg.predict_v2", "mmseg.metrics", "mmseg.flops_fps", "mmseg.draw", "mmseg.extract_loss_miou", ], "task_prefixes": ["mmseg."], "runtime_roles": ["mmseg_full"], "families": ["mmseg"], }, { "id": "visual", "label": "Visual Tools", "description": "可视化工具训练、推理、FPS、YOLO11 热度图、label 转换和 8-bit PNG 生成。", "required_tasks": [ "visual.train", "visual.inference", "visual.fps", "visual.yolo11_heatmap_v1", "visual.yolo11_heatmap_v2", "visual.deal_labels", ], "task_prefixes": ["visual."], "runtime_roles": ["backend_task"], "families": ["tool"], "evidence_roles": ["heatmap", "segmentation"], }, { "id": "analysis", "label": "Analysis", "description": "合并 SegModel/MMSeg/YOLO 指标，生成 CSV、表格、图表和性能摘要。", "required_tasks": ["analysis.all"], "task_prefixes": ["analysis."], "families": ["analysis"], }, { "id": "system", "label": "System", "description": "GPU 查询、环境检查、权重同步、任务日志/进度、取消和备份入口。", "required_tasks": ["system.backup", "system.check_graph_card"], "task_prefixes": ["system.", "mock."], "runtime_roles": ["backend_task", "mmseg_full"], }, ] def _tasks_for_group(all_tasks: list[str], prefixes: list[str]) -> list[str]: return sorted(task for task in all_tasks if any(task.startswith(prefix) for prefix in prefixes)) def _artifact_matches(item: dict[str, Any], group: dict[str, Any]) -> bool: families = set(group.get("families", [])) roles = set(group.get("evidence_roles", [])) return (not families or item.get("family") in families) and (not roles or item.get("role") in roles) def _coverage_build_task_set(coverage: dict[str, Any]) -> set[str]: return {item["task"] for item in coverage.get("task_build_checks", []) if item.get("passed")} def _runtime_map(readiness: dict[str, Any]) -> dict[str, dict[str, Any]]: return {item["role"]: item for item in readiness.get("envs", [])} def get_capability_matrix() -> dict[str, Any]: catalog = get_catalog() coverage = get_coverage_report() readiness = get_runtime_readiness(force=False) results = scan_results(limit=1000) curves = scan_training_curves(limit=50) datasets = list_uploaded_datasets() manifest = load_manifest() gpus = get_gpus() acceptance = latest_acceptance_report() real_acceptance = latest_real_acceptance_report() real_train_acceptance = latest_real_train_acceptance_report() deep_acceptance = latest_deep_acceptance_report() all_tasks = catalog["task_types"] buildable_tasks = _coverage_build_task_set(coverage) runtime_by_role = _runtime_map(readiness) real_train_curves = [item for item in curves if "real_train_acceptance" in item.get("relative_path", "")] domains = [] for group in CAPABILITY_GROUPS: group_tasks = _tasks_for_group(all_tasks, group["task_prefixes"]) required_tasks = group["required_tasks"] missing_required = [task for task in required_tasks if task not in all_tasks] unbuildable_required = [task for task in required_tasks if task in all_tasks and task not in buildable_tasks] runtime_roles = group.get("runtime_roles", []) runtime_reports = [runtime_by_role.get(role) for role in runtime_roles if runtime_by_role.get(role)] runtime_ready = all(item.get("passed") for item in runtime_reports) if runtime_roles else True artifacts = [item for item in results if _artifact_matches(item, group)] group_curves = [item for item in curves if not group.get("families") or item.get("family") in group["families"]] evidence_count = len(artifacts) + len(group_curves) if group["id"] == "dataset": evidence_count += sum(dataset["counts"]["images"] + dataset["counts"]["labels"] + dataset["counts"]["masks"] for dataset in datasets) if group["id"] == "mmseg": evidence_count += len(catalog.get("mmseg_algorithms", [])) if group["id"] == "system": evidence_count += int(bool(gpus.get("available"))) + int(manifest.get("count", 0) > 0) gaps = [] if missing_required: gaps.append(f"missing tasks: {', '.join(missing_required[:4])}") if unbuildable_required: gaps.append(f"unbuildable tasks: {', '.join(unbuildable_required[:4])}") if runtime_roles and not runtime_ready: gaps.append("runtime env check failed") domains.append( { "id": group["id"], "label": group["label"], "description": group["description"], "ready": not missing_required and not unbuildable_required and runtime_ready, "tasks": { "total": len(group_tasks), "required": len(required_tasks), "required_ready": len(required_tasks) - len(missing_required) - len(unbuildable_required), "examples": group_tasks[:10], "missing_required": missing_required, "unbuildable_required": unbuildable_required, }, "runtime": [ {"role": item["role"], "name": item["name"], "passed": item["passed"]} for item in runtime_reports ], "evidence": { "count": evidence_count, "artifacts": [ { "name": item["name"], "relative_path": item["relative_path"], "role": item.get("role"), "family": item.get("family"), } for item in artifacts[:6] ], "curves": [ { "name": item["name"], "relative_path": item["relative_path"], "family": item.get("family"), "row_count": item.get("row_count"), } for item in group_curves[:4] ], }, "gaps": gaps, } ) requirements = [ { "id": "user_script_mapping", "label": "用户侧脚本映射", "passed": coverage["mapped_user_scripts"] == coverage["user_scripts_total"] and not coverage["unmapped_user_scripts"], "detail": f"{coverage['mapped_user_scripts']}/{coverage['user_scripts_total']}", }, { "id": "runtime_readiness", "label": "运行环境就绪", "passed": readiness.get("passed", False), "detail": ", ".join(f"{item['name']}={item['passed']}" for item in readiness.get("envs", [])), }, { "id": "dataset_upload", "label": "数据集上传/Label/Mask", "passed": len(datasets) >= 1, "detail": f"{len(datasets)} uploaded dataset(s)", }, { "id": "yolo_heatmap", "label": "YOLO 热度图证据", "passed": any(item.get("family") == "yolo" and item.get("role") == "heatmap" for item in results), "detail": "heatmap artifact discovered", }, { "id": "training_curves", "label": "训练 loss/指标曲线", "passed": len(curves) >= 1, "detail": f"{len(curves)} curve file(s)", }, { "id": "real_train_curve", "label": "真实短训 loss 曲线", "passed": len(real_train_curves) >= 1, "detail": f"{len(real_train_curves)} real train curve file(s)", }, { "id": "deep_acceptance", "label": "深度训练验收", "passed": bool(deep_acceptance.get("passed")), "detail": deep_acceptance.get("run_id", "not run"), }, { "id": "real_workspace_acceptance", "label": "真实数据验收", "passed": bool(real_acceptance.get("passed")), "detail": real_acceptance.get("run_id", "not run"), }, { "id": "real_train_acceptance", "label": "真实短训练验收", "passed": bool(real_train_acceptance.get("passed")), "detail": real_train_acceptance.get("run_id", "not run"), }, { "id": "weights_manifest", "label": "权重清单", "passed": manifest.get("count", 0) >= 1, "detail": f"{manifest.get('count', 0)} weights indexed", }, ] ready_domains = sum(1 for item in domains if item["ready"]) return { "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "passed": ready_domains == len(domains) and all(item["passed"] for item in requirements), "summary": { "ready_domains": ready_domains, "total_domains": len(domains), "mapped_user_scripts": coverage["mapped_user_scripts"], "user_scripts_total": coverage["user_scripts_total"], "task_build_passed": coverage["task_build_passed"], "uploaded_datasets": len(datasets), "artifacts": len(results), "curves": len(curves), "real_train_curves": len(real_train_curves), "weights": manifest.get("count", 0), "gpus_available": bool(gpus.get("available")), "acceptance_passed": bool(acceptance.get("passed")), "real_acceptance_passed": bool(real_acceptance.get("passed")), "real_train_acceptance_passed": bool(real_train_acceptance.get("passed")), "deep_acceptance_passed": bool(deep_acceptance.get("passed")), }, "requirements": requirements, "domains": domains, }