diff --git a/README.md b/README.md index 200a728..f841f96 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,10 @@ are parsed into lightweight training curves. Job APIs and the SSE log stream also expose structured progress parsed from YOLO, MMSeg/MMEngine, SegModel-style epoch logs, and generic tqdm percentages, so the queue and live log panel can show stage, epoch/iteration, and percent -without changing the original training scripts. +without changing the original training scripts. Starting any web job or +dataset YOLO shortcut automatically opens its live log; the SSE stream resumes +from the current log size after the initial tail so existing lines are not +duplicated in the panel. The coverage panel calls `GET /api/coverage` and verifies that the user-facing scripts from the existing `Seg/` workspace are mapped to web jobs. MMSeg diff --git a/backend/app/agents/evaluation_agent.py b/backend/app/agents/evaluation_agent.py index d70bcdf..18833ed 100644 --- a/backend/app/agents/evaluation_agent.py +++ b/backend/app/agents/evaluation_agent.py @@ -59,6 +59,11 @@ def evaluate_project() -> dict: and "setResults(resultsNext)" in frontend_text and "slice(0, 240)" not in frontend_text, "job_progress_ui": "JobProgressBar" in frontend_text and "progressTrack" in frontend_text, + "live_log_stream_ui": "EventSource" in frontend_text + and "eventSourceRef" in frontend_text + and "log_size" in frontend_text + and "events?offset=" in frontend_text, + "live_log_offset_api": "log_size" in backend_text and "offset: int = Query(0, ge=0)" in backend_text, "runtime_readiness_ui": "runtimeReadiness" in frontend_text and "环境就绪" in frontend_text, "capability_matrix_ui": "capabilities" in frontend_text and "全功能矩阵" in frontend_text, "dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text, diff --git a/backend/app/main.py b/backend/app/main.py index f736cf2..aa27ae7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -39,6 +39,8 @@ 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 + log_path = Path(enriched["log_path"]) + enriched["log_size"] = log_path.stat().st_size if log_path.exists() else 0 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) @@ -183,9 +185,9 @@ def api_cancel_job(job_id: str) -> dict: @app.get("/api/jobs/{job_id}/events") -async def api_job_events(job_id: str): +async def api_job_events(job_id: str, offset: int = Query(0, ge=0)): async def stream(): - last_size = 0 + last_size = offset while True: job = db.get_job(job_id) if not job: @@ -196,6 +198,8 @@ async def api_job_events(job_id: str): chunk = "" if path.exists(): size = path.stat().st_size + if last_size > size: + last_size = size if size > last_size: with path.open("rb") as handle: handle.seek(last_size) diff --git a/backend/tests/test_jobs.py b/backend/tests/test_jobs.py index f30c4e3..2b45ff4 100644 --- a/backend/tests/test_jobs.py +++ b/backend/tests/test_jobs.py @@ -1,6 +1,38 @@ +from fastapi.testclient import TestClient + from app.jobs import default_conda_env_for_job +from app.main import _job_with_progress, app def test_mmseg_jobs_use_mmseg_conda_env_by_default(): assert default_conda_env_for_job("mmseg.train") == "seg_mmcv" assert default_conda_env_for_job("segmodel.train") == "seg_smp" + + +def test_job_progress_reports_log_size(tmp_path): + log_path = tmp_path / "job.log" + log_path.write_text("line one\nline two\n", encoding="utf-8") + job = {"id": "job1", "status": "running", "log_path": str(log_path)} + + enriched = _job_with_progress(job, include_log_tail=True) + + assert enriched["log_size"] == log_path.stat().st_size + assert enriched["log_tail"] == "line one\nline two\n" + + +def test_job_events_respect_log_offset(tmp_path, monkeypatch): + from app import main + + log_path = tmp_path / "job.log" + old_chunk = "old chunk\n" + log_path.write_text(old_chunk + "new chunk\n", encoding="utf-8") + + def fake_get_job(job_id): + return {"id": job_id, "status": "success", "log_path": str(log_path)} + + monkeypatch.setattr(main.db, "get_job", fake_get_job) + response = TestClient(app).get(f"/api/jobs/job1/events?offset={len(old_chunk.encode('utf-8'))}") + + assert response.status_code == 200 + assert "new chunk" in response.text + assert "old chunk" not in response.text diff --git a/frontend/src/main.tsx b/frontend/src/main.tsx index 7abbbdf..1486aab 100644 --- a/frontend/src/main.tsx +++ b/frontend/src/main.tsx @@ -1,4 +1,4 @@ -import React, { useEffect, useMemo, useState } from "react"; +import React, { useEffect, useMemo, useRef, useState } from "react"; import { createRoot } from "react-dom/client"; import { Activity, @@ -45,6 +45,7 @@ type Job = { started_at?: string; finished_at?: string; log_tail?: string; + log_size?: number; params: Record; progress?: JobProgress; }; @@ -430,6 +431,11 @@ function App() { const [uploadFiles, setUploadFiles] = useState(null); const [agentValidation, setAgentValidation] = useState(null); const [agentBusy, setAgentBusy] = useState(false); + const eventSourceRef = useRef(null); + + useEffect(() => () => { + eventSourceRef.current?.close(); + }, []); const runningCount = jobs.filter((job) => job.status === "running").length; const successCount = jobs.filter((job) => job.status === "success").length; @@ -515,10 +521,11 @@ function App() { async function createJob() { setBusy(true); try { - await api("/api/jobs", { + const job = await api("/api/jobs", { method: "POST", body: JSON.stringify({ type: taskType, params: JSON.parse(params) }) }); + await inspectJob(job); await refresh(); } finally { setBusy(false); @@ -633,13 +640,14 @@ function App() { try { const generated = await createSelectedYoloYaml(); if (!generated) return; - await api("/api/jobs", { + const job = await api("/api/jobs", { method: "POST", body: JSON.stringify({ type: "yolo.train_custom", params: { model: "YOLO11n-seg", data: generated.path, epochs: 10, imgsz: 640, batch: 1, workers: 0, device: "cpu", project: "var/custom_yolo_runs", name: selectedDataset.name, exist_ok: true } }) }); + await inspectJob(job); window.location.hash = "jobs"; await refresh(); } finally { @@ -651,13 +659,14 @@ function App() { if (!selectedDataset?.absolute_layout) return; setBusy(true); try { - await api("/api/jobs", { + const job = await api("/api/jobs", { method: "POST", body: JSON.stringify({ type: "yolo.predict_custom", params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, imgsz: 640, conf: 0.25, device: "cpu", project: "var/custom_yolo_runs", name: `${selectedDataset.name}_predict`, exist_ok: true } }) }); + await inspectJob(job); window.location.hash = "jobs"; await refresh(); } finally { @@ -669,13 +678,14 @@ function App() { if (!selectedDataset?.absolute_layout) return; setBusy(true); try { - await api("/api/jobs", { + const job = await api("/api/jobs", { method: "POST", body: JSON.stringify({ type: "yolo.heatmap_custom", params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, model_key: "YOLO11n-seg", cam_method: "GradCAM", target_layers: "model.model.model[9]", limit: 3, project: "var/custom_yolo_runs", name: `${selectedDataset.name}_heatmap` } }) }); + await inspectJob(job); window.location.hash = "jobs"; await refresh(); } finally { @@ -684,15 +694,24 @@ function App() { } async function inspectJob(job: Job) { + eventSourceRef.current?.close(); const detail = await api(`/api/jobs/${job.id}`); setSelectedJob(detail); setLog(detail.log_tail ?? ""); - const source = new EventSource(`${API_BASE}/api/jobs/${job.id}/events`); + const source = new EventSource(`${API_BASE}/api/jobs/${job.id}/events?offset=${detail.log_size ?? 0}`); + eventSourceRef.current = source; source.onmessage = (event) => { const payload = JSON.parse(event.data); if (payload.chunk) setLog((prev) => `${prev}${payload.chunk}`); setSelectedJob(payload.job); - if (["success", "failed", "cancelled"].includes(payload.job.status)) source.close(); + if (["success", "failed", "cancelled"].includes(payload.job.status)) { + source.close(); + if (eventSourceRef.current === source) eventSourceRef.current = null; + } + }; + source.onerror = () => { + source.close(); + if (eventSourceRef.current === source) eventSourceRef.current = null; }; }