Add result curve discovery dashboard

backend/app/modules/results/__init__.py

@@ -0,0 +1 @@
+"""Result artifact discovery and lightweight metric parsing."""

backend/app/modules/results/service.py

@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+import csv
+import math
+from pathlib import Path
+from typing import Iterable
+
+from ...config import settings
+
+RESULT_EXTS = {".csv", ".png", ".jpg", ".jpeg", ".svg", ".log", ".pth", ".pt", ".mp4", ".avi"}
+IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".svg"}
+CURVE_NAME_HINTS = ("results.csv", "loss", "metric", "miou", "iou")
+SERIES_HINTS = ("loss", "miou", "iou", "map", "precision", "recall", "acc", "dice", "f1")
+SKIP_SERIES_HINTS = ("time", "lr/")
+
+
+def result_roots() -> list[Path]:
+    source = settings.source_root
+    project = settings.project_root
+    roots = [
+        source / "DataSet_Public_outputs",
+        source / "BestMode_Predict_Results_DataSet_Public",
+        source / "Hardisk",
+        source / "Seg_All_In_One_Analysis",
+        source / "Seg_Predict_YoloModel",
+        source / "Seg_Predict_SegModel",
+        source / "Seg_Predict_Own_Video_V2",
+        source / "Tool-可视化" / "runs",
+        source / "Tool-可视化" / "Data" / "result",
+        source / "Tool-图片堆叠" / "result_0.3透明度",
+    ]
+    upload_root = project / "var" / "uploads" / "datasets"
+    if upload_root.exists():
+        roots.extend(path for path in upload_root.glob("*/results") if path.is_dir())
+    return roots
+
+
+def _safe_relative(path: Path) -> str:
+    resolved = path.resolve()
+    for root in (settings.source_root, settings.project_root):
+        try:
+            return str(resolved.relative_to(root))
+        except ValueError:
+            continue
+    return str(resolved)
+
+
+def _family_for_path(path: Path) -> str:
+    lower = str(path).lower()
+    if "yolo" in lower:
+        return "yolo"
+    if "mmseg" in lower or "dataset_public_outputs" in lower or "bestmode_predict" in lower:
+        return "mmseg"
+    if "segmodel" in lower:
+        return "segmodel"
+    if "analysis" in lower:
+        return "analysis"
+    if "tool-" in lower or "tool_" in lower:
+        return "tool"
+    return "artifact"
+
+
+def _role_for_path(path: Path) -> str:
+    lower = str(path).lower()
+    if any(key in lower for key in ("heat", "cam", "grad")):
+        return "heatmap"
+    if any(key in lower for key in ("predict", "pred", "mask", "comparison", "overlay", "result_")) and path.suffix.lower() in IMAGE_EXTS:
+        return "segmentation"
+    if any(key in lower for key in ("loss", "metric", "miou", "iou", "curve", "results.csv")):
+        return "curve"
+    if path.suffix.lower() in {".pt", ".pth"}:
+        return "weight"
+    if path.suffix.lower() in {".mp4", ".avi"}:
+        return "video"
+    return "artifact"
+
+
+def _iter_result_files() -> Iterable[Path]:
+    seen: set[Path] = set()
+    for root in result_roots():
+        if not root.exists():
+            continue
+        for path in root.rglob("*"):
+            if not path.is_file() or path.suffix.lower() not in RESULT_EXTS:
+                continue
+            resolved = path.resolve()
+            if resolved in seen:
+                continue
+            seen.add(resolved)
+            yield resolved
+
+
+def scan_results(limit: int = 1000) -> list[dict]:
+    results: list[dict] = []
+    for path in _iter_result_files():
+        try:
+            stat = path.stat()
+        except OSError:
+            continue
+        results.append(
+            {
+                "name": path.name,
+                "path": str(path),
+                "relative_path": _safe_relative(path),
+                "size": stat.st_size,
+                "modified": stat.st_mtime,
+                "kind": path.suffix.lower().lstrip("."),
+                "family": _family_for_path(path),
+                "role": _role_for_path(path),
+                "previewable": path.suffix.lower() in IMAGE_EXTS,
+            }
+        )
+    results.sort(key=lambda item: item["modified"], reverse=True)
+    return results[:limit]
+
+
+def _is_curve_candidate(path: Path) -> bool:
+    lower = path.name.lower()
+    whole = str(path).lower()
+    return path.suffix.lower() == ".csv" and (lower in CURVE_NAME_HINTS or any(hint in whole for hint in CURVE_NAME_HINTS))
+
+
+def _as_float(value: str | None) -> float | None:
+    if value is None:
+        return None
+    try:
+        number = float(value.strip())
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(number):
+        return None
+    return number
+
+
+def _series_candidates(headers: list[str]) -> list[str]:
+    candidates = []
+    for header in headers:
+        key = header.strip()
+        lower = key.lower()
+        if any(skip in lower for skip in SKIP_SERIES_HINTS):
+            continue
+        if any(hint in lower for hint in SERIES_HINTS):
+            candidates.append(key)
+    return candidates[:10]
+
+
+def parse_training_curve(path: Path, max_points: int = 300) -> dict | None:
+    try:
+        with path.open("r", encoding="utf-8-sig", newline="") as handle:
+            reader = csv.DictReader(handle)
+            if not reader.fieldnames:
+                return None
+            headers = [field.strip() for field in reader.fieldnames]
+            x_key = "epoch" if "epoch" in headers else "iter" if "iter" in headers else "step" if "step" in headers else headers[0]
+            series_names = _series_candidates(headers)
+            if not series_names:
+                return None
+            rows = []
+            for raw in reader:
+                normalized = {key.strip(): value for key, value in raw.items() if key is not None}
+                x = _as_float(normalized.get(x_key))
+                if x is None:
+                    continue
+                rows.append((x, normalized))
+    except (OSError, UnicodeDecodeError, csv.Error):
+        return None
+
+    if not rows:
+        return None
+    stride = max(1, math.ceil(len(rows) / max_points))
+    sampled = rows[::stride]
+    series = []
+    for name in series_names:
+        points = []
+        for x, row in sampled:
+            y = _as_float(row.get(name))
+            if y is not None:
+                points.append({"x": x, "y": y})
+        if len(points) >= 2:
+            values = [point["y"] for point in points]
+            series.append({"name": name, "points": points, "last": values[-1], "min": min(values), "max": max(values)})
+    if not series:
+        return None
+    stat = path.stat()
+    return {
+        "name": path.parent.name if path.name.lower() == "results.csv" else path.stem,
+        "file_name": path.name,
+        "path": str(path.resolve()),
+        "relative_path": _safe_relative(path),
+        "modified": stat.st_mtime,
+        "size": stat.st_size,
+        "family": _family_for_path(path),
+        "x_key": x_key,
+        "row_count": len(rows),
+        "series": series,
+    }
+
+
+def scan_training_curves(limit: int = 20) -> list[dict]:
+    curves = []
+    for path in _iter_result_files():
+        if not _is_curve_candidate(path):
+            continue
+        curve = parse_training_curve(path)
+        if curve:
+            curves.append(curve)
+    curves.sort(key=lambda item: item["modified"], reverse=True)
+    return curves[:limit]