File size: 3,445 Bytes
0185029 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | #!/usr/bin/env python3
"""Move legacy evaluation outputs into paper-level task families."""
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Iterable
TASK_FAMILIES = {
"video_qa": (
"videomme",
"videommev2",
"videommmu",
"mmvu",
"mvbench",
"videoholmes",
"longvideobench",
"lvbench",
"mlvu",
),
"spatial_intelligence": ("vsi", "mmsi", "mindcube", "revsi"),
"spatial_temporal_grounding": ("stvg",),
}
def _exists(path: Path) -> bool:
return path.exists() or path.is_symlink()
def planned_moves(root: Path) -> list[tuple[Path, Path]]:
moves = []
for family, tasks in TASK_FAMILIES.items():
for task in tasks:
source = root / task
if source.is_dir():
moves.append((source, root / family / task))
return moves
def _preflight_merge(source: Path, destination: Path) -> None:
for child in source.iterdir():
target = destination / child.name
if not _exists(target):
continue
if child.is_dir() and not child.is_symlink():
if not target.is_dir() or target.is_symlink():
raise FileExistsError(f"cannot merge directory into {target}")
_preflight_merge(child, target)
continue
raise FileExistsError(f"refusing to overwrite existing output: {target}")
def _merge(source: Path, destination: Path) -> None:
destination.mkdir(parents=True, exist_ok=True)
for child in source.iterdir():
target = destination / child.name
if child.is_dir() and not child.is_symlink() and target.is_dir():
_merge(child, target)
else:
child.rename(target)
source.rmdir()
def organize(root: Path, moves: Iterable[tuple[Path, Path]]) -> None:
moves = list(moves)
for source, destination in moves:
if _exists(destination):
_preflight_merge(source, destination)
for source, destination in moves:
destination.parent.mkdir(parents=True, exist_ok=True)
if _exists(destination):
_merge(source, destination)
else:
source.rename(destination)
def create_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"root",
type=Path,
help="One model's output root, for example outputs/Video-ORA-9B.",
)
parser.add_argument(
"--apply",
action="store_true",
help="Perform the moves. Without this flag, only print the plan.",
)
return parser
def main() -> int:
args = create_parser().parse_args()
root = args.root.expanduser().resolve()
if not root.is_dir():
raise SystemExit(f"output root is not a directory: {root}")
moves = planned_moves(root)
if not moves:
print(f"Already organized: {root}")
return 0
action = "MOVE" if args.apply else "PLAN"
for source, destination in moves:
print(f"{action} {source.relative_to(root)} -> {destination.relative_to(root)}")
if args.apply:
organize(root, moves)
print(f"Organized {len(moves)} task director{'y' if len(moves) == 1 else 'ies'}.")
else:
print("Dry run only; pass --apply to move these directories.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|