#!/usr/bin/env python
"""SAM 후처리 작업 모니터 HTML 을 만든다.
수치는 전부 prompt_stats.py 의 집계 JSON 에서 온다. 이 파일은 숫자를 만들지
않는다. 단계·이슈·설정 같은 사람이 쓰는 것만 상태 JSON 에서 읽는다.
python tools/make_merge_monitor.py \
--stats output/sam3/YYX_run/_prompt_stats.json \
--status configs/sam_merge_status.json \
--out docs/pilot/sam_merge.html
"""
import argparse
import html
import json
import os
import time
from collections import defaultdict
STATE = {
"done": ("완료", "#1a7f4b", "#e3f5ea"),
"running": ("진행", "#8a5a00", "#fdf0d5"),
"blocked": ("막힘", "#a12a2a", "#fbe6e6"),
"todo": ("대기", "#5a5f66", "#eceef0"),
"open": ("미해결", "#a12a2a", "#fbe6e6"),
"closed": ("해결", "#1a7f4b", "#e3f5ea"),
"rejected": ("기각", "#5a5f66", "#eceef0"),
}
def esc(s):
return html.escape(str(s))
def badge(state):
label, fg, bg = STATE.get(state, (state, "#5a5f66", "#eceef0"))
return (f''
f'{esc(label)}')
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--stats", required=True, help="병합 전 집계 JSON")
ap.add_argument("--status", required=True, help="단계·이슈 JSON")
ap.add_argument("--stats-after", help="병합 후 집계 JSON (--pattern '*_merged.json')")
ap.add_argument("--coverage", help="coverage_stats.py 집계 JSON (무라벨 지표)")
ap.add_argument("--theater", nargs="*", default=[], metavar="제목=경로",
help="공정을 보여줄 뷰어 HTML. 경로는 --out 기준 상대경로")
ap.add_argument("--out", required=True)
args = ap.parse_args()
with open(args.stats, encoding="utf-8") as fh:
st = json.load(fh)
with open(args.status, encoding="utf-8") as fh:
sv = json.load(fh)
af = None
if args.stats_after:
with open(args.stats_after, encoding="utf-8") as fh:
af = json.load(fh)
cv = None
if args.coverage:
with open(args.coverage, encoding="utf-8") as fh:
cv = json.load(fh)
stages = []
for t in args.theater:
if "=" not in t:
raise SystemExit(f"--theater 형식은 '제목=경로': {t}")
name, path = t.split("=", 1)
if not os.path.isfile(os.path.join(os.path.dirname(os.path.abspath(args.out)), path)):
raise SystemExit(f"뷰어 없음: {path}")
stages.append((name, path))
rows = st["rows"]
gsum = defaultdict(lambda: {"poly": 0, "area": 0.0, "prompts": 0, "used": 0})
for r in rows:
g = gsum[r["group"]]
g["poly"] += r["polygons"]
g["area"] += r["area_px"]
g["prompts"] += 1
g["used"] += 1 if r["polygons"] else 0
tot_area = st["total_area_px"] or 1
tot_poly = st["total_polygons"] or 1
parts = []
A = parts.append
A(f'{esc(sv["title"])}')
A("""""")
A('
')
A(f'
{esc(sv["title"])}
')
A(f'
{esc(sv.get("note", ""))}
')
A(f'
생성 {time.strftime("%Y-%m-%d %H:%M")} · '
f'집계 {esc(os.path.basename(st["run_dir"]))}
')
A('
실측
')
cards = [
("사진", f'{st["photos"]:,}'),
("폴리곤", f'{st["total_polygons"]:,}'),
("라벨 종류", f'{sum(1 for r in rows if r["polygons"]):,} / {len(rows):,}'),
("집합", f'{sum(1 for g in gsum if g != "없음"):,}'),
("검출 면적 / 화면", f'{100 * tot_area / st["image_area_px"]:.1f}%'
if st.get("image_area_px") else "-"),
("집합 없는 폴리곤",
f'{gsum["없음"]["poly"]:,}' if "없음" in gsum else "0"),
]
if af:
cards.insert(2, ("병합 후 폴리곤",
f'{af["total_polygons"]:,} '
f'(-{100 * (tot_poly - af["total_polygons"]) / tot_poly:.0f}%)'))
for k, v in cards:
A(f'
{esc(k)}
{esc(v)}
')
A('
')
if cv:
per = sorted(cv["per_photo"], key=lambda r: r["covered"])
un = [100 * (1 - r["covered"]) for r in per]
A('
')
for k, v in [("무라벨 평균", f'{100 * (1 - cv["covered_mean"]):.1f}%'),
("무라벨 중앙", f'{sorted(un)[len(un) // 2]:.1f}%'),
("최악", f'{100 * (1 - cv["covered_min"]):.1f}%'),
("5% 이하 사진", f'{sum(1 for u in un if u <= 5)} / {len(un)}')]:
A(f'
{esc(k)}
'
f'
{esc(v)}
')
A('
')
A('
'
'
사진
폴리곤
무라벨%
')
for r in per:
u = 100 * (1 - r["covered"])
A(f'
{esc(r["photo"].replace("_multi.json", ""))}
'
f'
{r["segments"]:,}
'
f'
{u:.1f}%
'
f'
5 else "#1a7f4b"}">
')
A('
')
if stages:
A('
공정 보기 — 같은 사진, 단계별
')
if af:
A(f'
SAM 검출 {st["total_polygons"]:,} 폴리곤'
f'({sum(1 for r in st["rows"] if r["polygons"])} 라벨)'
f'→ 집합 병합 →'
f'{af["total_polygons"]:,} 폴리곤'
f'({sum(1 for r in af["rows"] if r["polygons"])} 집합)'
f'· 사진 {st["photos"]}장 기준
')
A('
')
for i, (name, _) in enumerate(stages):
A(f'')
A('
')
for i, (_, path) in enumerate(stages):
A(f'
'
f'
')
A('
왼쪽 목록에서 라벨을 켜고 끈다. '
'휠 확대, 드래그 이동. 새 창: '
+ " · ".join(f'{esc(n)}' for n, p in stages) + '
')
A('
단계
')
for s in sv.get("steps", []):
A(f'
{esc(s["id"])}{badge(s["state"])}'
f'
{esc(s["name"])}
'
f'
{esc(s.get("detail", ""))}
')
A('
이슈 · 기각된 시도
')
for i in sv.get("issues", []):
A(f'
{badge(i["state"])}
{esc(i["text"])}
')
apoly, aarea = defaultdict(int), defaultdict(float)
if af:
for r in af["rows"]:
apoly[r["group"]] += r["polygons"]
aarea[r["group"]] += r["area_px"]
atot = af["total_area_px"] or 1
A('
집합별 합계
'
'
집합
프롬프트(검출>0/전체)
폴리곤
폴리곤%
'
'
면적%
')
if af:
A('
병합 후 폴리곤
줄어든 비율
병합 후 면적%
')
A('
')
for g, v in sorted(gsum.items(), key=lambda t: -t[1]["poly"]):
share = 100 * v["area"] / tot_area
A(f'
{esc(g)}
{v["used"]} / {v["prompts"]}
'
f'
{v["poly"]:,}
{100 * v["poly"] / tot_poly:.1f}%
'
f'
{share:.2f}%
')
if af:
n = apoly.get(g, 0)
drop = 100 * (v["poly"] - n) / v["poly"] if v["poly"] else 0
cls = "d-up" if n < v["poly"] else ""
A(f'
{n:,}
{drop:.0f}%
'
f'
{100 * aarea.get(g, 0.0) / atot:.2f}%
')
A(f'
'
f'
')
A('
')
if af:
A('
병합 후 면적%는 병합 후 총검출면적 기준. '
'겹침을 빼지 않은 다각형 넓이 합이다.
')
cats = sorted({r["category"] for r in rows})
groups = sorted({r["group"] for r in rows})
A('
카테고리 · 프롬프트별 검출
')
A('
'
''
''
''
''
'
')
A('
'
'
프롬프트
카테고리
집합
폴리곤
사진
'
'
면적%
중앙 면적 px
점수 중앙
')
for r in rows:
cls = []
if not r["polygons"]:
cls.append("zero")
if r["group"] == "없음":
cls.append("nogroup")
c = f' class="{" ".join(cls)}"' if cls else ""
dash = "-"
A(f''
f'
{esc(r["prompt"])}
{esc(r["category"])}
'
f'
{esc(r["group"])}
'
f'
{r["polygons"]:,}
'
f'
{r["photos"] or dash}
'
f'
'
f'{100 * r["area_share"]:.2f}%
'
f'
'
f'{r["area_median_px"]:,.0f}
'
f'
'
f'{r["score_median"]:.3f}
')
A('
')
if sv.get("params"):
A('
설정
')
for k, v in sv["params"].items():
A(f'{esc(k)}{esc(v)}')
A('
')
A('
')
A("""""")
os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
with open(args.out, "w", encoding="utf-8") as fh:
fh.write("\n".join(parts))
print(f"저장: {args.out} ({os.path.getsize(args.out):,} bytes)")
if __name__ == "__main__":
main()