""" Railway Detection Web UI 사용법: python tools/web_ui.py 브라우저: http://localhost:7000 """ import asyncio import base64 import json import os import queue import subprocess import sys import threading import uuid from pathlib import Path import cv2 import numpy as np try: from fastapi import FastAPI from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse import uvicorn except ImportError: print("FastAPI/uvicorn 설치 필요: pip install fastapi uvicorn") sys.exit(1) app = FastAPI() jobs: dict = {} # job_id -> {queue, status, result_path} ROOT = Path(__file__).parent.parent # 프로젝트 루트 # ── 미리보기 이미지 생성 ─────────────────────────────────────────────────────── def make_preview(image_path: str, cols: int, rows: int, selected_rows: list) -> str: buf = np.fromfile(image_path, dtype=np.uint8) img = cv2.imdecode(buf, cv2.IMREAD_COLOR) if img is None: raise ValueError(f"이미지 로드 실패: {image_path}") H, W = img.shape[:2] tw = 2000 scale = tw / W vis = cv2.resize(img, (tw, int(H * scale))) H_s, W_s = vis.shape[:2] tile_w = W_s / cols tile_h = H_s / rows for r in range(rows): for c in range(cols): idx = r * cols + c + 1 x0, y0 = int(c * tile_w), int(r * tile_h) x1, y1 = min(W_s, int((c + 1) * tile_w)), min(H_s, int((r + 1) * tile_h)) row_num = r + 1 if row_num in selected_rows: overlay = vis.copy() cv2.rectangle(overlay, (x0, y0), (x1, y1), (0, 200, 0), -1) cv2.addWeighted(overlay, 0.25, vis, 0.75, 0, vis) cv2.rectangle(vis, (x0, y0), (x1, y1), (0, 255, 0), 2) else: cv2.rectangle(vis, (x0, y0), (x1, y1), (180, 180, 180), 1) label = str(idx) fs = max(0.35, tile_w / 120) (tw2, th2), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fs, 1) cx, cy = (x0 + x1) // 2, (y0 + y1) // 2 cv2.putText(vis, label, (cx - tw2 // 2, cy + th2 // 2), cv2.FONT_HERSHEY_SIMPLEX, fs, (255, 255, 255), 1, cv2.LINE_AA) # Row 번호 y0 = int(r * tile_h) row_num = r + 1 color = (0, 255, 0) if row_num in selected_rows else (160, 160, 160) cv2.putText(vis, f"Row{row_num}", (4, y0 + 18), cv2.FONT_HERSHEY_SIMPLEX, 0.55, color, 2, cv2.LINE_AA) _, enc = cv2.imencode(".jpg", vis, [cv2.IMWRITE_JPEG_QUALITY, 88]) return base64.b64encode(enc).decode("utf-8") def tiles_from_rows(selected_rows: list, cols: int) -> str: if not selected_rows: return "all" ids = [] for r in sorted(selected_rows): ids.extend(range((r - 1) * cols + 1, r * cols + 1)) if ids and ids[-1] - ids[0] + 1 == len(ids): return f"{ids[0]}-{ids[-1]}" return ",".join(str(t) for t in ids) # ── API ─────────────────────────────────────────────────────────────────────── @app.post("/api/preview") async def api_preview(data: dict): try: b64 = make_preview( data["image_path"], data.get("cols", 8), data.get("rows", 6), data.get("selected_rows", []), ) return {"ok": True, "image": b64} except Exception as e: return {"ok": False, "error": str(e)} @app.post("/api/detect") async def api_detect(data: dict): job_id = str(uuid.uuid4())[:8] q: queue.Queue = queue.Queue() jobs[job_id] = {"queue": q, "status": "running", "result_path": None, "proc": None} def run(): tiles_str = tiles_from_rows(data.get("selected_rows", []), data.get("cols", 8)) cmd = [ sys.executable, "-u", str(ROOT / "tools" / "detect_all_objects.py"), "--input", data["image_path"], "--categories", data.get("categories", "configs/railway_zone.json"), "--tiles", tiles_str, "--cols", str(data.get("cols", 8)), "--rows", str(data.get("rows", 6)), "--overlap", str(data.get("overlap", 0.20)), "--conf", str(data.get("conf", 0.20)), "--workers", str(data.get("workers", 8)), "--save-json" ] env = {**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"} flags = subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0 try: proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", env=env, cwd=str(ROOT), creationflags=flags, ) jobs[job_id]["proc"] = proc for raw in proc.stdout: for line in raw.replace("\r", "\n").split("\n"): line = line.strip() if not line: continue q.put(("log", line)) if line.startswith("저장:"): jobs[job_id]["result_path"] = line.replace("저장:", "").strip() proc.wait() if jobs[job_id]["status"] == "stopped": q.put(("error", "사용자가 중지했습니다")) elif proc.returncode == 0: q.put(("done", jobs[job_id]["result_path"] or "완료")) jobs[job_id]["status"] = "done" else: q.put(("error", f"종료코드 {proc.returncode}")) jobs[job_id]["status"] = "error" except Exception as e: q.put(("error", str(e))) jobs[job_id]["status"] = "error" threading.Thread(target=run, daemon=True).start() return {"job_id": job_id} @app.post("/api/stop/{job_id}") async def api_stop(job_id: str): job = jobs.get(job_id) if not job: return {"ok": False, "error": "not found"} job["status"] = "stopped" proc = job.get("proc") if proc and proc.poll() is None: if os.name == "nt": subprocess.call(["taskkill", "/F", "/T", "/PID", str(proc.pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) else: proc.terminate() return {"ok": True} @app.get("/api/progress/{job_id}") async def api_progress(job_id: str): if job_id not in jobs: return JSONResponse({"error": "not found"}, status_code=404) async def event_gen(): q = jobs[job_id]["queue"] while True: try: type_, msg = q.get_nowait() yield f"data: {json.dumps({'type': type_, 'msg': msg})}\n\n" if type_ in ("done", "error"): break except queue.Empty: await asyncio.sleep(0.25) yield ": ping\n\n" return StreamingResponse(event_gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) @app.get("/api/result/{job_id}") async def api_result(job_id: str): job = jobs.get(job_id, {}) path = job.get("result_path") if not path or not Path(path).exists(): return {"ok": False, "error": "결과 없음"} buf = np.fromfile(path, dtype=np.uint8) img = cv2.imdecode(buf, cv2.IMREAD_COLOR) if img is None: return {"ok": False, "error": "이미지 로드 실패"} h, w = img.shape[:2] if max(h, w) > 3000: s = 3000 / max(h, w) img = cv2.resize(img, (int(w * s), int(h * s))) _, enc = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 90]) return {"ok": True, "image": base64.b64encode(enc).decode(), "path": path} @app.post("/api/open-folder") async def api_open_folder(data: dict): path = Path(data.get("path", "output/detect")) folder = path.parent if path.is_file() else path if os.name == "nt": os.startfile(str(folder.resolve())) return {"ok": True} # ── HTML ────────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def index(): return HTML HTML = r""" Railway Detection UI

🛤 Railway Detection UI

휠: 확대/축소  |  드래그: 이동 이미지 경로 입력 후 미리보기 클릭
결과 없음 휠: 확대/축소  |  드래그: 이동
검출 완료 후 결과가 표시됩니다
대기 중
""" if __name__ == "__main__": print("Railway Detection UI 시작: http://localhost:7000") uvicorn.run(app, host="0.0.0.0", port=7000, log_level="warning")