import argparse import csv import json import re import sys from pathlib import Path NUMBER_RE = r"[-+]?(?:\d+(?:\.\d*)?|\.\d+)" FIELD_RE = re.compile(rf"([A-Za-z0-9_]+)\s*:\s*({NUMBER_RE})") ITER_RE = re.compile(r"iteration\s+(\d+)") BEST_AVG_RE = re.compile(rf"val_best_avg_dice:\s*({NUMBER_RE})\s+at\s+(\d+)\s+iter") NAMESPACE_RE = re.compile(r"Namespace\((.*)\)") ARG_RE = re.compile(r"--([A-Za-z0-9_]+)(?:\s+([^-]\S*))?") SUMMARY_COLUMNS = [ "path", "dataset", "save_name", "model", "lb_domain", "lb_num", "domain_num", "max_iterations", "last_iteration", "last_epoch", "unet_best_avg_dice", "unet_best_avg_iter", "sam_best_avg_dice", "sam_best_avg_iter", "unet_final_avg_dice", "sam_final_avg_dice", "last_loss", "last_mask_ratio", ] def parse_namespace(line): match = NAMESPACE_RE.search(line) if not match: return {} values = {} for part in split_top_level(match.group(1)): if "=" not in part: continue key, value = part.split("=", 1) values[key.strip()] = parse_value(value.strip()) return values def split_top_level(text): parts = [] current = [] quote = None depth = 0 for char in text: if quote: current.append(char) if char == quote: quote = None continue if char in {"'", '"'}: quote = char current.append(char) continue if char in "([{": depth += 1 elif char in ")]}": depth -= 1 if char == "," and depth == 0: parts.append("".join(current).strip()) current = [] else: current.append(char) if current: parts.append("".join(current).strip()) return parts def parse_value(value): if len(value) >= 2 and value[0] in {"'", '"'} and value[-1] == value[0]: return value[1:-1] if value in {"True", "False"}: return value == "True" if value == "None": return None try: return int(value) except ValueError: pass try: return float(value) except ValueError: return value def parse_command(line): if "python train.py" not in line: return {} values = {} for key, value in ARG_RE.findall(line): values[key] = True if value == "" else value return values def parse_metric_fields(line): return {key: float(value) for key, value in FIELD_RE.findall(line)} def average_dice(metrics): dice_values = [ value for key, value in metrics.items() if key.startswith("val_") and key.endswith("_dice") ] if not dice_values: return None return sum(dice_values) / len(dice_values) def summarize_log(path, model_root): path = Path(path) model_root = Path(model_root) row = { "path": str(path.relative_to(model_root)) if path.is_relative_to(model_root) else str(path), "dataset": None, "save_name": None, "model": None, "lb_domain": None, "lb_num": None, "domain_num": None, "max_iterations": None, "last_iteration": None, "last_epoch": None, "unet_best_avg_dice": None, "unet_best_avg_iter": None, "sam_best_avg_dice": None, "sam_best_avg_iter": None, "unet_final_avg_dice": None, "sam_final_avg_dice": None, "last_loss": None, "last_mask_ratio": None, } current_model = None pending_epoch_model = None pending_epoch_metrics = None with path.open("r", encoding="utf-8", errors="replace") as handle: for raw_line in handle: line = raw_line.strip() if not line: continue command_values = parse_command(line) namespace_values = parse_namespace(line) for values in (command_values, namespace_values): for key in ("dataset", "save_name", "model", "lb_domain", "lb_num", "domain_num", "max_iterations"): if key in values: row[key] = values[key] if "iteration " in line: iter_match = ITER_RE.search(line) if iter_match: row["last_iteration"] = int(iter_match.group(1)) fields = parse_metric_fields(line) row["last_loss"] = fields.get("loss", row["last_loss"]) row["last_mask_ratio"] = fields.get("mask_ratio", row["last_mask_ratio"]) if "test unet model" in line: pending_epoch_model = None pending_epoch_metrics = None current_model = "unet" continue if "test sam model" in line: pending_epoch_model = None pending_epoch_metrics = None current_model = "sam" continue if " epoch " in line and "domain" not in line: epoch_match = re.search(r"epoch\s+(\d+)", line) if epoch_match: row["last_epoch"] = int(epoch_match.group(1)) pending_epoch_model = current_model pending_epoch_metrics = {} continue if pending_epoch_model and line.startswith("val_"): pending_epoch_metrics.update(parse_metric_fields(line)) final_avg = average_dice(pending_epoch_metrics) if final_avg is not None: row[f"{pending_epoch_model}_final_avg_dice"] = final_avg continue if pending_epoch_model: pending_epoch_model = None pending_epoch_metrics = None if "val_best_avg_dice:" in line: best_match = BEST_AVG_RE.search(line) if best_match and current_model: prefix = "sam" if current_model == "sam" else "unet" row[f"{prefix}_best_avg_dice"] = float(best_match.group(1)) row[f"{prefix}_best_avg_iter"] = int(best_match.group(2)) return row def find_logs(model_dir): return sorted(Path(model_dir).glob("**/log.txt")) def write_csv(rows, output): writer = csv.DictWriter(output, fieldnames=SUMMARY_COLUMNS) writer.writeheader() for row in rows: writer.writerow({key: format_value(row.get(key)) for key in SUMMARY_COLUMNS}) def write_json(rows, output): json.dump(rows, output, indent=2) output.write("\n") def write_markdown(rows, output): output.write("| " + " | ".join(SUMMARY_COLUMNS) + " |\n") output.write("| " + " | ".join(["---"] * len(SUMMARY_COLUMNS)) + " |\n") for row in rows: output.write("| " + " | ".join(format_value(row.get(key)) for key in SUMMARY_COLUMNS) + " |\n") def write_summary(rows, output, output_format): if output_format == "csv": write_csv(rows, output) elif output_format == "json": write_json(rows, output) else: write_markdown(rows, output) def format_value(value): if value is None: return "" if isinstance(value, float): return f"{value:.6f}" return str(value) def parse_args(argv): parser = argparse.ArgumentParser( description="Summarize training results from model/**/log.txt files." ) parser.add_argument( "--model-dir", type=Path, default=Path("../model"), help="Directory that contains dataset experiment folders. Default: ../model", ) parser.add_argument( "--format", choices=("markdown", "csv", "json"), default="markdown", help="Output format. Default: markdown", ) parser.add_argument( "--output", type=Path, help="Write the summary to this file instead of printing it to stdout.", ) return parser.parse_args(argv) def main(argv=None): args = parse_args(sys.argv[1:] if argv is None else argv) logs = find_logs(args.model_dir) rows = [summarize_log(path, args.model_dir) for path in logs] if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) with args.output.open("w", encoding="utf-8", newline="") as output: write_summary(rows, output, args.format) else: write_summary(rows, sys.stdout, args.format) return 0 if __name__ == "__main__": raise SystemExit(main())