admin/Seg_Data_Server_Net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add deep training acceptance checks

Browse Source
This commit is contained in:
admin
2026-06-30 13:42:30 +08:00
parent 7d6e1692b1
commit cf920e97c3
8 changed files with 179 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
backend/app/acceptance.py
View File

@@ -62,6 +62,66 @@ MMSEG_FULL_BUILD_SNIPPET = (
)
SEGMODEL_TRAIN_STEP_SNIPPET = (
"import torch, segmentation_models_pytorch as smp; "
"torch.manual_seed(7); "
"model=smp.Unet(encoder_name='resnet18', encoder_weights=None, classes=2).train(); "
"inputs=torch.randn(2,3,64,64); "
"targets=torch.randint(0,2,(2,64,64)); "
"optimizer=torch.optim.SGD(model.parameters(), lr=1e-3); "
"outputs=model(inputs); "
"loss=torch.nn.functional.cross_entropy(outputs, targets); "
"loss.backward(); optimizer.step(); "
"print('loss', round(float(loss.detach()), 6), 'shape', tuple(outputs.shape))"
)
def _yolo_tiny_train_snippet(root: Path, weight: Path) -> str:
return (
"import shutil, cv2, numpy as np; "
"from pathlib import Path; "
"from ultralytics import YOLO; "
f"root=Path({str(root)!r}); weight={str(weight)!r}; "
"shutil.rmtree(root, ignore_errors=True); "
"[ (root / item).mkdir(parents=True, exist_ok=True) for item in ['images/train','images/val','labels/train','labels/val','runs'] ]; "
"image=np.zeros((64,64,3), dtype=np.uint8); "
"cv2.rectangle(image, (16,16), (48,48), (255,255,255), -1); "
"label='0 0.25 0.25 0.75 0.25 0.75 0.75 0.25 0.75\\n'; "
"\nfor split in ['train','val']:\n"
" cv2.imwrite(str(root / 'images' / split / 'sample.jpg'), image)\n"
" (root / 'labels' / split / 'sample.txt').write_text(label, encoding='utf-8')\n"
"(root / 'data.yaml').write_text('path: '+str(root)+'\\ntrain: images/train\\nval: images/val\\nnc: 1\\nnames:\\n 0: object\\n', encoding='utf-8'); "
"model=YOLO(weight); "
"model.train(data=str(root/'data.yaml'), epochs=1, imgsz=64, batch=1, workers=0, device='cpu', project=str(root/'runs'), name='tiny', exist_ok=True, verbose=False, plots=False, val=False); "
"results=root/'runs'/'tiny'/'results.csv'; best=root/'runs'/'tiny'/'weights'/'best.pt'; "
"assert results.exists() and results.stat().st_size > 0; "
"assert best.exists() and best.stat().st_size > 0; "
"print('results', results, results.stat().st_size, 'best', best.stat().st_size)"
)
def _mmseg_train_step_snippet(config_path: Path) -> str:
return (
"import torch; "
"from mmengine.config import Config; "
"from mmengine.structures import PixelData; "
"from mmseg.registry import MODELS; "
"from mmseg.structures import SegDataSample; "
"from mmseg.utils import register_all_modules; "
"register_all_modules(init_default_scope=True); "
f"cfg=Config.fromfile({str(config_path)!r}); "
"cfg.model.backbone.init_cfg=None; cfg.model.pretrained=None; "
"model=MODELS.build(cfg.model).train(); "
"sample=SegDataSample(); "
"sample.gt_sem_seg=PixelData(data=torch.randint(0,19,(1,64,64), dtype=torch.long)); "
"losses=model(torch.randn(1,3,64,64), [sample], mode='loss'); "
"loss=sum(value if torch.is_tensor(value) else sum(value) for value in losses.values()); "
"optimizer=torch.optim.SGD(model.parameters(), lr=1e-4); "
"loss.backward(); optimizer.step(); "
"print('loss', round(float(loss.detach()), 6), sorted(losses.keys()))"
)
def _request_json(method: str, url: str, payload: dict[str, Any] | None = None, timeout: int = 10) -> dict[str, Any]:
data = None
headers = {"Accept": "application/json"}
@@ -252,6 +312,56 @@ def latest_acceptance_report() -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def latest_deep_acceptance_report() -> dict[str, Any]:
path = settings.project_root / "var" / "acceptance" / "deep_latest.json"
if not path.exists():
return {"available": False, "path": str(path)}
return json.loads(path.read_text(encoding="utf-8"))
def run_deep_acceptance() -> dict[str, Any]:
"""Run minimal training loops for each model family without full datasets."""
acceptance_root = settings.project_root / "var" / "acceptance"
run_id = uuid.uuid4().hex[:8]
fixture_root = acceptance_root / f"deep_{run_id}"
fixture_root.mkdir(parents=True, exist_ok=True)
yolo_weight = settings.source_root / "Seg_All_In_One_YoloModel" / "yolo11n-seg.pt"
mmseg_config = settings.source_root / "Seg_All_In_One_MMSeg" / "configs" / "fcn" / "fcn_r18-d8_4xb2-80k_cityscapes-512x1024.py"
checks = [
{
"name": "segmodel_tiny_train_step",
"passed": False,
"detail": _run_snippet(SEGMODEL_TRAIN_STEP_SNIPPET, timeout=90),
},
{
"name": "yolo_tiny_segment_train_epoch",
"passed": False,
"detail": _run_snippet(_yolo_tiny_train_snippet(fixture_root / "yolo_tiny", yolo_weight), timeout=180),
},
{
"name": "mmseg_tiny_train_step",
"passed": False,
"detail": _run_conda_snippet(settings.mmseg_conda_env, _mmseg_train_step_snippet(mmseg_config), timeout=120),
},
]
for check in checks:
check["passed"] = bool(check["detail"].get("passed"))
report = {
"available": True,
"run_id": run_id,
"fixture_root": str(fixture_root),
"passed": all(item["passed"] for item in checks),
"checks": checks,
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
latest = acceptance_root / "deep_latest.json"
latest.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
return report
def run_live_acceptance(base_url: str = "http://127.0.0.1:8010") -> dict[str, Any]:
"""Run a lightweight end-to-end smoke against the live API and job runner."""
acceptance_root = settings.project_root / "var" / "acceptance"

2
backend/app/agents/evaluation_agent.py
View File

@@ -42,6 +42,8 @@ def evaluate_project() -> dict:
"loss_result_ui": "loss" in frontend_text.lower() and "heatmap" in frontend_text.lower() and "CurvePanel" in frontend_text,
"dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text,
"curve_api": "/api/results/curves" in backend_text,
"deep_acceptance_api": "/api/acceptance/deep" in backend_text,
"deep_acceptance_ui": "runDeepAcceptance" in frontend_text and "深度训练" in frontend_text,
"coverage_api": "/api/coverage" in backend_text and coverage["task_build_passed"],
"visual_tools": "visual.yolo11_heatmap_v2" in catalog["task_types"] and "visual.fps" in catalog["task_types"],
"yolo_dataset_tools": "dataset.yolo_txt_sort" in catalog["task_types"] and "dataset.yolo_resize" in catalog["task_types"],

5
backend/app/agents/validation_agent.py
View File

@@ -8,7 +8,7 @@ import urllib.error
import urllib.request
from pathlib import Path
from ..acceptance import run_live_acceptance
from ..acceptance import run_deep_acceptance, run_live_acceptance
from ..catalog import get_catalog
from ..config import settings
from ..coverage import get_coverage_report
@@ -108,6 +108,9 @@ def validate_project(run_build: bool = False) -> dict:
if os.getenv("SEG_VALIDATE_ACCEPTANCE", "1") == "1":
acceptance = run_live_acceptance(backend_url)
checks.append({"name": "live_acceptance_smoke", "passed": acceptance["passed"], "detail": acceptance})
if os.getenv("SEG_VALIDATE_DEEP", "1") == "1":
deep_acceptance = run_deep_acceptance()
checks.append({"name": "deep_training_acceptance", "passed": deep_acceptance["passed"], "detail": deep_acceptance})
if run_build:
tests = _run(["conda", "run", "-n", settings.backend_conda_env, "python", "-m", "pytest", "-q"], cwd=settings.project_root, timeout=120)

12
backend/app/main.py
View File

@@ -9,7 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, StreamingResponse
from . import db
from .acceptance import latest_acceptance_report, run_live_acceptance
from .acceptance import latest_acceptance_report, latest_deep_acceptance_report, run_deep_acceptance, run_live_acceptance
from .catalog import get_catalog
from .config import settings
from .coverage import get_coverage_report
@@ -81,6 +81,16 @@ def api_acceptance_smoke(base_url: str = "http://127.0.0.1:8010") -> dict:
return run_live_acceptance(base_url)
@app.get("/api/acceptance/deep/latest")
def api_deep_acceptance_latest() -> dict:
return latest_deep_acceptance_report()
@app.post("/api/acceptance/deep")
def api_deep_acceptance() -> dict:
return run_deep_acceptance()
@app.get("/api/datasets")
def api_datasets() -> list[dict]:
return list_uploaded_datasets()