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

Add structured job progress tracking

Browse Source
This commit is contained in:
admin
2026-06-30 14:18:02 +08:00
parent 93af8bcd3a
commit 442b521705
8 changed files with 454 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -42,8 +42,10 @@ def evaluate_project() -> dict:
"upload_ui": "uploadDatasetFiles" in frontend_text and "labels" in frontend_text and "masks" in frontend_text,
"dataset_quality_ui": "DatasetQuality" in frontend_text and "generateSelectedYoloYaml" in frontend_text,
"loss_result_ui": "loss" in frontend_text.lower() and "heatmap" in frontend_text.lower() and "CurvePanel" in frontend_text,
"job_progress_ui": "JobProgressBar" in frontend_text and "progressTrack" in frontend_text,
"dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text,
"dataset_quality_api": "/api/datasets/{dataset_name}/validate" in backend_text and "/api/datasets/{dataset_name}/yolo-yaml" in backend_text,
"job_progress_api": "progress_from_log_path" in backend_text and '"progress"' 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,

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

@@ -15,6 +15,7 @@ from ..coverage import get_coverage_report
from ..modules.results.service import scan_training_curves
from ..modules.system.service import get_conda_envs, get_gpus
from ..modules.weights.service import load_manifest
from ..progress import parse_progress
def _run(command: list[str], cwd: Path | None = None, timeout: int = 60) -> dict:
@@ -57,6 +58,8 @@ def validate_project(run_build: bool = False) -> dict:
checks.append({"name": "weights_manifest_present", "passed": manifest.get("count", 0) >= 1})
curves = scan_training_curves()
checks.append({"name": "training_curves_detected", "passed": len(curves) >= 1, "detail": {"count": len(curves), "examples": [item["relative_path"] for item in curves[:5]]}})
parsed_progress = parse_progress("Epoch(train) [2][25/50] lr: 1e-3\n", status="running")
checks.append({"name": "job_progress_parser", "passed": parsed_progress["stage"] == "training" and parsed_progress["percent"] == 50, "detail": parsed_progress})
checks.append({"name": "gpus_query", "passed": bool(get_gpus().get("available"))})
env_names = [item["name"] for item in get_conda_envs().get("envs", [])]
checks.append({"name": "task_env_exists", "passed": settings.task_conda_env in env_names, "detail": {"env": settings.task_conda_env}})
@@ -98,11 +101,21 @@ def validate_project(run_build: bool = False) -> dict:
frontend_url = os.getenv("SEG_VALIDATE_FRONTEND_URL", "http://127.0.0.1:5173")
health = _fetch(f"{backend_url}/api/health")
datasets = _fetch(f"{backend_url}/api/datasets")
live_jobs = _fetch(f"{backend_url}/api/jobs")
live_coverage = _fetch(f"{backend_url}/api/coverage")
live_curves = _fetch(f"{backend_url}/api/results/curves")
frontend = _fetch(frontend_url)
checks.append({"name": "live_backend_health", "passed": health["passed"] and '"ok":true' in health.get("body", "").replace(" ", ""), "detail": health})
checks.append({"name": "live_dataset_api", "passed": datasets["passed"] and datasets.get("body", "").lstrip().startswith("["), "detail": datasets})
try:
live_job_items = json.loads(live_jobs.get("body", "[]")) if live_jobs.get("passed") else []
except Exception:
live_job_items = []
checks.append({
"name": "live_jobs_progress_api",
"passed": live_jobs["passed"] and isinstance(live_job_items, list) and (not live_job_items or "progress" in live_job_items[0]),
"detail": live_jobs,
})
checks.append({"name": "live_coverage_api", "passed": live_coverage["passed"] and '"task_build_passed":true' in live_coverage.get("body", "").replace(" ", ""), "detail": live_coverage})
checks.append({"name": "live_training_curves_api", "passed": live_curves["passed"] and live_curves.get("body", "").lstrip().startswith("["), "detail": live_curves})
checks.append({"name": "live_frontend_index", "passed": frontend["passed"] and "Seg Data Server" in frontend.get("body", ""), "detail": frontend})

20
backend/app/main.py
View File

@@ -21,6 +21,7 @@ from .modules.weights.service import load_manifest, sync_weights, verify_weights
from .agents.evaluation_agent import evaluate_project
from .agents.validation_agent import validate_project
from .paths import ensure_inside
from .progress import progress_from_log_path
from .schemas import DatasetCreate, DatasetYoloYamlRequest, JobCreate, ProfileCreate, WeightSyncRequest
app = FastAPI(title="Seg Data Server Net", version="0.1.0")
@@ -34,6 +35,15 @@ app.add_middleware(
)
def _job_with_progress(job: dict, include_log_tail: bool = False) -> dict:
enriched = dict(job)
max_bytes = 65536 if include_log_tail else 32768
enriched["progress"] = progress_from_log_path(enriched["log_path"], enriched["status"], max_bytes=max_bytes)
if include_log_tail:
enriched["log_tail"] = db.log_tail(enriched["log_path"], max_bytes=max_bytes)
return enriched
@app.on_event("startup")
def startup() -> None:
db.init_db()
@@ -135,14 +145,14 @@ def api_save_profile(profile: ProfileCreate) -> dict:
@app.post("/api/jobs")
def api_create_job(request: JobCreate) -> dict:
try:
return create_job(request)
return _job_with_progress(create_job(request))
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@app.get("/api/jobs")
def api_jobs(limit: int = 100) -> list[dict]:
return db.list_jobs(limit)
return [_job_with_progress(job) for job in db.list_jobs(limit)]
@app.get("/api/jobs/{job_id}")
@@ -150,8 +160,7 @@ def api_job(job_id: str) -> dict:
job = db.get_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="job not found")
job["log_tail"] = db.log_tail(job["log_path"])
return job
return _job_with_progress(job, include_log_tail=True)
@app.post("/api/jobs/{job_id}/cancel")
@@ -159,7 +168,7 @@ def api_cancel_job(job_id: str) -> dict:
job = cancel_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="job not found")
return job
return _job_with_progress(job)
@app.get("/api/jobs/{job_id}/events")
@@ -171,6 +180,7 @@ async def api_job_events(job_id: str):
if not job:
yield "event: error\ndata: job not found\n\n"
return
job = _job_with_progress(job)
path = Path(job["log_path"])
chunk = ""
if path.exists():

247
backend/app/progress.py Normal file
View File

@@ -0,0 +1,247 @@
from __future__ import annotations
import re
from pathlib import Path
from typing import Any
from . import db
ANSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
ULTRALYTICS_EPOCH_RE = re.compile(r"^\s*(?P<current>\d+)\s*/\s*(?P<total>\d+)(?=\s)")
EPOCH_WORD_RE = re.compile(
r"\bepoch\b[^0-9]{0,12}(?P<current>\d+)\s*(?:/|of)\s*(?P<total>\d+)",
re.IGNORECASE,
)
MMENGINE_EPOCH_RE = re.compile(
r"Epoch\((?P<stage>train|val|test)\)\s*\[(?P<epoch>\d+)\]\s*\[(?P<current>\d+)\s*/\s*(?P<total>\d+)\]",
re.IGNORECASE,
)
MMENGINE_ITER_RE = re.compile(
r"Iter\((?P<stage>train|val|test)\)\s*\[(?P<current>\d+)\s*/\s*(?P<total>\d+)\]",
re.IGNORECASE,
)
TQDM_PERCENT_RE = re.compile(r"(?P<percent>\d{1,3}(?:\.\d+)?)%\|")
COMMAND_EPOCHS_RE = re.compile(r"(?:--epochs|--max-epochs|max_epochs)\D{0,20}(?P<total>\d+)", re.IGNORECASE)
def strip_ansi(value: str) -> str:
return ANSI_RE.sub("", value.replace("\r", "\n"))
def progress_from_log_path(log_path: str | Path, status: str | None = None, max_bytes: int = 65536) -> dict[str, Any]:
return parse_progress(db.log_tail(log_path, max_bytes=max_bytes), status=status)
def parse_progress(log_text: str, status: str | None = None) -> dict[str, Any]:
text = strip_ansi(log_text or "")
parsed = _parse_training_progress(text)
if parsed is None:
parsed = _status_progress(status)
if status == "success":
return {
**parsed,
"percent": 100.0,
"stage": "completed",
"label": _terminal_label("已完成", parsed),
"source": "status",
}
if status == "failed":
return {
**parsed,
"stage": "failed",
"label": _terminal_label("失败", parsed),
"source": parsed.get("source", "status"),
}
if status == "cancelled":
return {
**parsed,
"stage": "cancelled",
"label": _terminal_label("已取消", parsed),
"source": parsed.get("source", "status"),
}
return parsed
def _parse_training_progress(text: str) -> dict[str, Any] | None:
lines = [line.strip() for line in text.splitlines() if line.strip()]
if not lines:
return None
total_epochs = _find_total_epochs(text)
for line in reversed(lines[-400:]):
mmengine_epoch = MMENGINE_EPOCH_RE.search(line)
if mmengine_epoch:
epoch = int(mmengine_epoch.group("epoch"))
current = int(mmengine_epoch.group("current"))
total = int(mmengine_epoch.group("total"))
stage = _stage_name(mmengine_epoch.group("stage"))
if total_epochs:
percent = ((max(epoch - 1, 0) + current / max(total, 1)) / total_epochs) * 100
return _progress(
percent=percent,
label=f"Epoch {epoch}/{total_epochs} · iter {current}/{total}",
stage=stage,
current=epoch,
total=total_epochs,
unit="epoch",
source="mmengine_epoch",
)
return _progress(
percent=(current / max(total, 1)) * 100,
label=f"Epoch {epoch} · iter {current}/{total}",
stage=stage,
current=current,
total=total,
unit="iter",
source="mmengine_iter_in_epoch",
)
epoch_word = EPOCH_WORD_RE.search(line)
if epoch_word:
current = int(epoch_word.group("current"))
total = int(epoch_word.group("total"))
return _progress(
percent=(current / max(total, 1)) * 100,
label=f"Epoch {current}/{total}",
stage=_stage_from_line(line),
current=current,
total=total,
unit="epoch",
source="epoch_word",
)
ultralytics_epoch = ULTRALYTICS_EPOCH_RE.search(line)
if ultralytics_epoch:
current = int(ultralytics_epoch.group("current"))
total = int(ultralytics_epoch.group("total"))
if total > 1 or "loss" in line.lower() or "gpu" in line.lower():
return _progress(
percent=(current / max(total, 1)) * 100,
label=f"Epoch {current}/{total}",
stage=_stage_from_line(line),
current=current,
total=total,
unit="epoch",
source="ultralytics_epoch",
)
mmengine_iter = MMENGINE_ITER_RE.search(line)
if mmengine_iter:
current = int(mmengine_iter.group("current"))
total = int(mmengine_iter.group("total"))
return _progress(
percent=(current / max(total, 1)) * 100,
label=f"Iter {current}/{total}",
stage=_stage_name(mmengine_iter.group("stage")),
current=current,
total=total,
unit="iter",
source="mmengine_iter",
)
tqdm = TQDM_PERCENT_RE.search(line)
if tqdm:
percent = float(tqdm.group("percent"))
if 0 <= percent <= 100:
return _progress(
percent=percent,
label=f"{percent:g}%",
stage=_stage_from_line(line),
current=None,
total=None,
unit="percent",
source="tqdm_percent",
)
tail = "\n".join(lines[-20:]).lower()
if "results saved" in tail or "save_dir=" in tail:
return _progress(
percent=99.0,
label="保存结果",
stage="saving",
current=None,
total=None,
unit=None,
source="completion_hint",
)
if "starting training" in tail or "train:" in tail:
return _progress(
percent=None,
label="训练中",
stage="training",
current=None,
total=total_epochs,
unit="epoch" if total_epochs else None,
source="training_hint",
)
return None
def _find_total_epochs(text: str) -> int | None:
values = [int(match.group("total")) for match in COMMAND_EPOCHS_RE.finditer(text)]
return max(values) if values else None
def _status_progress(status: str | None) -> dict[str, Any]:
if status == "queued":
return _progress(0.0, "排队中", "queued", None, None, None, "status")
if status == "running":
return _progress(None, "运行中", "running", None, None, None, "status")
if status == "success":
return _progress(100.0, "已完成", "completed", None, None, None, "status")
if status == "failed":
return _progress(None, "失败", "failed", None, None, None, "status")
if status == "cancelled":
return _progress(None, "已取消", "cancelled", None, None, None, "status")
return _progress(None, "未开始", "unknown", None, None, None, "status")
def _progress(
percent: float | None,
label: str,
stage: str,
current: int | None,
total: int | None,
unit: str | None,
source: str,
) -> dict[str, Any]:
if percent is not None:
percent = round(max(0.0, min(100.0, float(percent))), 2)
return {
"percent": percent,
"label": label,
"stage": stage,
"current": current,
"total": total,
"unit": unit,
"source": source,
}
def _terminal_label(prefix: str, parsed: dict[str, Any]) -> str:
label = parsed.get("label")
if label and label not in {prefix, "运行中", "排队中", "未开始"}:
return f"{prefix} · {label}"
return prefix
def _stage_from_line(line: str) -> str:
lower = line.lower()
if "val" in lower or "validat" in lower:
return "validation"
if "predict" in lower or "infer" in lower:
return "prediction"
if "save" in lower:
return "saving"
return "training"
def _stage_name(value: str) -> str:
lower = value.lower()
if lower == "val":
return "validation"
if lower == "test":
return "evaluation"
return "training"

60
backend/tests/test_progress.py Normal file
View File

@@ -0,0 +1,60 @@
from pathlib import Path
from app.progress import parse_progress, progress_from_log_path
def test_parse_ultralytics_epoch_progress():
progress = parse_progress(
"""
Epoch GPU_mem box_loss seg_loss cls_loss dfl_loss Instances Size
3/10 0G 1.234 2.345 0.456 0.789 2 640: 100%|##########| 2/2
""",
status="running",
)
assert progress["percent"] == 30
assert progress["current"] == 3
assert progress["total"] == 10
assert progress["unit"] == "epoch"
assert progress["stage"] == "training"
def test_parse_mmengine_epoch_uses_total_epochs_from_command():
progress = parse_progress(
"""
COMMAND: python tools/train.py config.py --max-epochs 5
Epoch(train) [2][25/50] lr: 1.0000e-03 eta: 0:00:10
""",
status="running",
)
assert progress["percent"] == 30
assert progress["label"] == "Epoch 2/5 · iter 25/50"
assert progress["current"] == 2
assert progress["total"] == 5
def test_parse_tqdm_percent_fallback():
progress = parse_progress("Predict: 42%|#### | 21/50 [00:01<00:02]\n", status="running")
assert progress["percent"] == 42
assert progress["unit"] == "percent"
assert progress["stage"] == "prediction"
def test_terminal_success_overrides_percent_but_keeps_context():
progress = parse_progress("Epoch 2/4 loss=0.5\n", status="success")
assert progress["percent"] == 100
assert progress["stage"] == "completed"
assert progress["label"] == "已完成 · Epoch 2/4"
def test_progress_from_log_path_reads_tail(tmp_path: Path):
log_path = tmp_path / "train.log"
log_path.write_text("Epoch 1/2 loss=0.8\n", encoding="utf-8")
progress = progress_from_log_path(log_path, status="running")
assert progress["percent"] == 50
assert progress["source"] == "epoch_word"