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透明度", project / "var" / "custom_yolo_runs", ] 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 points: 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]