215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
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())
|
|
acceptance_root = project / "var" / "acceptance"
|
|
if acceptance_root.exists():
|
|
roots.extend(path for path in acceptance_root.glob("deep_*/segmodel_tiny") if path.is_dir())
|
|
roots.extend(path for path in acceptance_root.glob("deep_*/mmseg_tiny") if path.is_dir())
|
|
roots.extend(path for path in acceptance_root.glob("deep_*/yolo_tiny/runs/tiny") if path.is_dir())
|
|
roots.extend(path for path in acceptance_root.glob("deep_*/yolo_tiny/runs/tiny/HeartMap_Visual") 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]
|