1716 lines
70 KiB
Python
1716 lines
70 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
from datetime import datetime, timezone
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import numpy as np
|
|
|
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
|
from fastapi import Response
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
import math
|
|
from pydantic import BaseModel, field_validator
|
|
import config as app_config
|
|
import uvicorn
|
|
|
|
from config import (
|
|
IS_DEV,
|
|
MAX_RENDER_POINTS,
|
|
POINT_CLOUD_INITIAL_DENSITY_PERCENT,
|
|
POINT_CLOUD_INITIAL_FILTER,
|
|
POINT_CLOUD_INITIAL_POINT_SIZE,
|
|
TERRAIN_MODEL_CONFIG,
|
|
)
|
|
|
|
# 유틸리티 연동
|
|
from utils.structurizer import structurize_las
|
|
from utils.filter_grid_min_z import filter_grid_min_z
|
|
from utils.filter_csf import filter_csf
|
|
from utils.filter_pmf import filter_pmf
|
|
from utils.filter_ransac import filter_ransac
|
|
from utils.terrain_model_converter import MODEL_METHODS, build_all_terrain_models
|
|
from utils.vworld_downloader import download_vworld_satellite_map
|
|
from utils.download_gis_vectors import download_all_gis_vectors
|
|
from utils.mvt_helper import generate_mvt_tile
|
|
|
|
|
|
app = FastAPI(title="Forest Road Ground Filtering System", version="1.0.0")
|
|
|
|
_ACTIVE_SECTION_BUILDS: set[str] = set()
|
|
_ACTIVE_SECTION_BUILDS_GUARD = threading.Lock()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# 폴더 설정
|
|
ROOT_DIR = Path(__file__).resolve().parent
|
|
INSTANCE_DIR = ROOT_DIR / "instance"
|
|
SAMPLE_DIR = ROOT_DIR / "samples" / "step1_scan"
|
|
FRONTEND_DIST = ROOT_DIR / "frontend" / "dist"
|
|
FRONTEND_ASSETS = FRONTEND_DIST / "assets"
|
|
|
|
INSTANCE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
if FRONTEND_ASSETS.exists():
|
|
app.mount("/assets", StaticFiles(directory=FRONTEND_ASSETS), name="frontend-assets")
|
|
|
|
|
|
POINT_CLOUD_FILTER_KEYS = {"grid_min_z", "csf", "pmf", "ransac"}
|
|
|
|
|
|
def get_point_cloud_viewer_config() -> dict[str, int | float | str]:
|
|
initial_filter = (
|
|
POINT_CLOUD_INITIAL_FILTER
|
|
if POINT_CLOUD_INITIAL_FILTER in POINT_CLOUD_FILTER_KEYS
|
|
else "csf"
|
|
)
|
|
return {
|
|
"initial_density_percent": POINT_CLOUD_INITIAL_DENSITY_PERCENT,
|
|
"initial_point_size": POINT_CLOUD_INITIAL_POINT_SIZE,
|
|
"initial_filter": initial_filter,
|
|
"max_render_points": MAX_RENDER_POINTS,
|
|
}
|
|
|
|
|
|
@app.get("/api/config")
|
|
def get_app_config() -> dict[str, object]:
|
|
"""프론트엔드가 런타임에 사용할 공개 설정을 반환합니다."""
|
|
return {
|
|
"is_dev": IS_DEV,
|
|
"point_cloud": get_point_cloud_viewer_config(),
|
|
"terrain_model": TERRAIN_MODEL_CONFIG,
|
|
"route": {
|
|
"default_grade_class": app_config.ROUTE_DEFAULT_GRADE_CLASS,
|
|
"grade_classes": list(app_config.ROUTE_GRADE_CLASSES),
|
|
"avoid_default_radius_m": app_config.ROUTE_AVOID_DEFAULT_RADIUS_M,
|
|
"avoid_radius_range_m": list(app_config.ROUTE_AVOID_RADIUS_RANGE_M),
|
|
"min_curve_radius_range_m": list(app_config.ROUTE_MIN_CURVE_RADIUS_RANGE_M),
|
|
"grade_limit_range": list(app_config.ROUTE_GRADE_LIMIT_RANGE),
|
|
"min_curve_radius_by_class": app_config.FOREST_ROAD_MIN_CURVE_R_M,
|
|
"max_grade_by_class": app_config.FOREST_ROAD_MAX_GRADE,
|
|
"max_grade_paved": app_config.ROUTE_MAX_GRADE_PAVED,
|
|
"required_point_tolerance_m": app_config.ROUTE_REQUIRED_POINT_TOLERANCE_M,
|
|
},
|
|
}
|
|
|
|
|
|
def ensure_terrain_models(proj_dir: Path, *, force: bool = False) -> dict[str, Any]:
|
|
"""기존 프로젝트 임시 폴더에서 15개 지표면 모델 캐시를 생성하거나 재사용합니다."""
|
|
npz_path = proj_dir / "structured.npz"
|
|
if not npz_path.exists():
|
|
raise FileNotFoundError("structured.npz가 없어 지표면 모델을 생성할 수 없습니다.")
|
|
ground_masks: dict[str, np.ndarray] = {}
|
|
for source_filter in TERRAIN_MODEL_CONFIG["source_filters"]:
|
|
mask_path = proj_dir / f"mask_{source_filter}.npy"
|
|
if not mask_path.exists():
|
|
raise FileNotFoundError(f"{mask_path.name}이 없어 지표면 모델을 생성할 수 없습니다.")
|
|
ground_masks[source_filter] = np.load(mask_path, mmap_mode="r")
|
|
with np.load(npz_path) as structured_data:
|
|
return build_all_terrain_models(
|
|
structured_data,
|
|
ground_masks,
|
|
proj_dir / "terrain_models",
|
|
TERRAIN_MODEL_CONFIG,
|
|
force=force,
|
|
)
|
|
|
|
|
|
# ---- API 스키마 ----
|
|
class CheckSampleRequest(BaseModel):
|
|
filenames: list[str]
|
|
|
|
|
|
# ---- 핵심 오케스트레이션 로직 ----
|
|
def process_pipeline(project_id: str, las_path: Path):
|
|
"""프로젝트 업로드 또는 샘플 트리거 시 최초 1회 전체 파이프라인 수행.
|
|
|
|
모든 진행률은 터미널의 한 줄 안에서 % 단위로 실시간 통합 업데이트됩니다.
|
|
연산 속도가 빠른 단계에서도 숫자가 연속적으로 자연스럽게 올라가는 보간 처리가 지원됩니다.
|
|
"""
|
|
import sys
|
|
import threading
|
|
total_start = time.time()
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 통합 상태 관리용 딕셔너리
|
|
progress = {
|
|
"구조화": 0,
|
|
"Min_Z": 0,
|
|
"CSF": 0,
|
|
"PMF": 0,
|
|
"RANSAC": 0
|
|
}
|
|
|
|
# 보간 스레드 제어용 변수
|
|
current_stage = None
|
|
stop_stage_event = threading.Event()
|
|
|
|
def print_progress_line():
|
|
"""진행 상태를 터미널 한 줄에 갱신 출력하는 헬퍼 함수"""
|
|
line = " | ".join(f"{k}: {v}%" for k, v in progress.items())
|
|
sys.stdout.write(f"\r[진행] {line}")
|
|
sys.stdout.flush()
|
|
|
|
def smooth_progress_worker():
|
|
"""데몬 스레드: 지정된 단계의 수치를 95%까지 부드럽게 연속 증가시킴"""
|
|
while True:
|
|
# 10ms 단위 대기
|
|
if stop_stage_event.wait(timeout=0.03):
|
|
# 이벤트가 세팅되면 중지 신호이므로 재대기
|
|
stop_stage_event.clear()
|
|
continue
|
|
|
|
stage = current_stage
|
|
if stage and progress[stage] < 95:
|
|
# 1%씩 연속적으로 올리기
|
|
progress[stage] += 1
|
|
print_progress_line()
|
|
|
|
# 보간 스레드 기동
|
|
t_interpolator = threading.Thread(target=smooth_progress_worker, daemon=True)
|
|
t_interpolator.start()
|
|
|
|
print(f"\n[{project_id}] 파이프라인 선행 연산 시작 (LAS: {las_path.name})")
|
|
print_progress_line()
|
|
|
|
# 1. 구조화 단계
|
|
current_stage = "구조화"
|
|
def struct_cb(pct):
|
|
# 콜백 진행률이 현재 보간값보다 크면 업데이트
|
|
if pct > progress["구조화"]:
|
|
progress["구조화"] = pct
|
|
print_progress_line()
|
|
|
|
npz_path = structurize_las(las_path, proj_dir, progress_callback=struct_cb)
|
|
stop_stage_event.set()
|
|
current_stage = None
|
|
progress["구조화"] = 100
|
|
print_progress_line()
|
|
|
|
# 구조화 데이터 로딩
|
|
data = np.load(npz_path)
|
|
n_total = len(data["xyz"])
|
|
|
|
# 2. Grid Min-Z 필터 단계
|
|
current_stage = "Min_Z"
|
|
# 실제 필터는 0.2~0.5초만에 끝나버리므로, 보간 스레드가 숫자를 연속으로 올리도록 잠시 시차를 두고 수행
|
|
time.sleep(0.15)
|
|
mask_min_z = filter_grid_min_z(data)
|
|
np.save(proj_dir / "mask_grid_min_z.npy", mask_min_z)
|
|
stop_stage_event.set()
|
|
current_stage = None
|
|
progress["Min_Z"] = 100
|
|
print_progress_line()
|
|
|
|
# 3. CSF 필터 단계
|
|
current_stage = "CSF"
|
|
time.sleep(0.15)
|
|
try:
|
|
mask_csf = filter_csf(data)
|
|
except Exception:
|
|
mask_csf = np.zeros(n_total, dtype=bool)
|
|
np.save(proj_dir / "mask_csf.npy", mask_csf)
|
|
stop_stage_event.set()
|
|
current_stage = None
|
|
progress["CSF"] = 100
|
|
print_progress_line()
|
|
|
|
# 4. PMF 필터 단계 (시간이 약 1~2초가량 다소 소요됨)
|
|
current_stage = "PMF"
|
|
try:
|
|
mask_pmf = filter_pmf(data, proj_dir)
|
|
except Exception:
|
|
mask_pmf = np.zeros(n_total, dtype=bool)
|
|
np.save(proj_dir / "mask_pmf.npy", mask_pmf)
|
|
stop_stage_event.set()
|
|
current_stage = None
|
|
progress["PMF"] = 100
|
|
print_progress_line()
|
|
|
|
# 5. RANSAC 필터 단계
|
|
current_stage = "RANSAC"
|
|
def ransac_cb(pct):
|
|
if pct > progress["RANSAC"]:
|
|
progress["RANSAC"] = pct
|
|
print_progress_line()
|
|
|
|
try:
|
|
mask_ransac = filter_ransac(data, progress_callback=ransac_cb)
|
|
except Exception:
|
|
mask_ransac = np.zeros(n_total, dtype=bool)
|
|
np.save(proj_dir / "mask_ransac.npy", mask_ransac)
|
|
stop_stage_event.set()
|
|
current_stage = None
|
|
progress["RANSAC"] = 100
|
|
print_progress_line()
|
|
|
|
# 6. 개발 검증용 지표면 모델 15개 조합 사전 계산
|
|
sys.stdout.write("\n")
|
|
build_all_terrain_models(
|
|
data,
|
|
{"grid_min_z": mask_min_z, "csf": mask_csf, "pmf": mask_pmf},
|
|
proj_dir / "terrain_models",
|
|
TERRAIN_MODEL_CONFIG,
|
|
)
|
|
|
|
# 7. VWorld 위성사진 및 백지도, 하이브리드 지도 타일 다운로드 및 병합 (방안 B 및 확장)
|
|
prj_candidate = proj_dir / "result.prj"
|
|
|
|
bounds = {
|
|
"x": [float(data["bounds"][0][0]), float(data["bounds"][0][1])],
|
|
"y": [float(data["bounds"][1][0]), float(data["bounds"][1][1])],
|
|
"z": [float(data["bounds"][2][0]), float(data["bounds"][2][1])],
|
|
}
|
|
|
|
# 3개 타입의 지도를 각각 다운로드
|
|
layers = [
|
|
{"layer": "Satellite", "ext": "jpeg"},
|
|
{"layer": "Hybrid", "ext": "png"},
|
|
{"layer": "white", "ext": "png"}
|
|
]
|
|
|
|
for item in layers:
|
|
layer_name = item["layer"]
|
|
ext = item["ext"]
|
|
print(f"\n[{project_id}] VWorld {layer_name} 지도 다운로드 시작...")
|
|
try:
|
|
download_vworld_satellite_map(prj_candidate, bounds, proj_dir, layer_name=layer_name, ext=ext)
|
|
print(f"[{project_id}] VWorld {layer_name} 다운로드 및 메타데이터 작성 완료!")
|
|
except Exception as e:
|
|
print(f"[{project_id}] VWorld {layer_name} 다운로드 실패: {e}")
|
|
|
|
# 8. 국가 GIS 표준 벡터 데이터 다운로드 및 저장 (지적도, 용도지역, 행정구역, 수계망, 산사태위험등급)
|
|
print(f"\n[{project_id}] 국가 GIS 표준 벡터 데이터(GeoJSON) 다운로드 시작...")
|
|
try:
|
|
download_all_gis_vectors(prj_candidate, bounds, proj_dir)
|
|
print(f"[{project_id}] 국가 GIS 표준 벡터 데이터 수집 완료!")
|
|
except Exception as e:
|
|
print(f"[{project_id}] 국가 GIS 벡터 데이터 수집 실패: {e}")
|
|
|
|
total_duration = time.time() - total_start
|
|
# 완료 개행 및 요약 정보 출력
|
|
sys.stdout.write("\n")
|
|
print(f"[{project_id}] 전체 필터 선행 연산 완료! (총 소요 시간: {total_duration:.2f}s)\n")
|
|
|
|
|
|
|
|
# ---- REST API 엔드포인트 ----
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/preview", response_model=None)
|
|
def get_project_preview(project_id: str) -> Response:
|
|
"""프로젝트 지형 영역의 2D preview 이미지를 반환합니다.
|
|
TIF 원본 파일이 있을 경우 고해상도 PNG로 변환하여 캐싱 파일(preview.png)로 저장한 뒤 서빙합니다.
|
|
"""
|
|
import io
|
|
from PIL import Image
|
|
import rasterio
|
|
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
cached_png = proj_dir / "preview.png"
|
|
|
|
# 이미 캐싱된 PNG 파일이 존재하는 경우 즉시 리턴 (메모리 및 CPU 절약)
|
|
if cached_png.exists():
|
|
return FileResponse(cached_png, media_type="image/png")
|
|
|
|
raw_dir = proj_dir / "raw"
|
|
tif_candidates = []
|
|
if raw_dir.exists():
|
|
tif_candidates = list(raw_dir.glob("*.tif")) + list(raw_dir.glob("*.tiff"))
|
|
|
|
if tif_candidates:
|
|
preview_path = tif_candidates[0]
|
|
else:
|
|
raise HTTPException(status_code=404, detail="업로드된 원본 TIF 미리보기 파일이 존재하지 않습니다.")
|
|
|
|
try:
|
|
# 최초 1회 변환 및 파일 저장 캐싱
|
|
with rasterio.open(preview_path) as src:
|
|
if src.count >= 3:
|
|
r = src.read(1)
|
|
g = src.read(2)
|
|
b = src.read(3)
|
|
def normalize(band):
|
|
b_min, b_max = band.min(), band.max()
|
|
if b_max > b_min:
|
|
return ((band - b_min) / (b_max - b_min) * 255).astype('uint8')
|
|
return band.astype('uint8')
|
|
rgb = np.dstack((normalize(r), normalize(g), normalize(b)))
|
|
img = Image.fromarray(rgb)
|
|
else:
|
|
band = src.read(1)
|
|
b_min, b_max = band.min(), band.max()
|
|
if b_max > b_min:
|
|
gray = ((band - b_min) / (b_max - b_min) * 255).astype('uint8')
|
|
else:
|
|
gray = band.astype('uint8')
|
|
img = Image.fromarray(gray)
|
|
|
|
max_size = 3000
|
|
if img.width > max_size or img.height > max_size:
|
|
img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
|
|
|
|
img.save(cached_png, format="PNG")
|
|
|
|
return FileResponse(cached_png, media_type="image/png")
|
|
except Exception as e:
|
|
if cached_png.exists():
|
|
try:
|
|
cached_png.unlink()
|
|
except Exception:
|
|
pass
|
|
raise HTTPException(status_code=500, detail=f"TIF 파일 변환 실패: {e}")
|
|
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/vworld-meta")
|
|
def get_vworld_meta(project_id: str, layer_name: str = "satellite") -> dict[str, Any]:
|
|
"""VWorld 위성 맵 이미지 매핑 좌표 메타데이터를 반환합니다.
|
|
layer_name: satellite, hybrid, white, gray, midnight
|
|
"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
|
|
# gray 요청 시 white 레이어 파일로 우회 매핑
|
|
target_layer = "white" if layer_name.lower() in ["gray", "white"] else layer_name.lower()
|
|
meta_name = f"vworld_{target_layer}_meta.json"
|
|
meta_path = proj_dir / meta_name
|
|
|
|
if not meta_path.exists() and target_layer == "satellite":
|
|
meta_path = proj_dir / "vworld_meta.json"
|
|
|
|
if not meta_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"VWorld {layer_name} (매핑: {target_layer}) 메타데이터를 찾을 수 없습니다.")
|
|
try:
|
|
with open(meta_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"메타데이터 로드 실패: {e}")
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/vworld-map")
|
|
def get_vworld_map(project_id: str, layer_name: str = "satellite") -> FileResponse:
|
|
"""배경 지도 레이어 PNG 이미지를 반환합니다.
|
|
layer_name: satellite, hybrid, white, gray, midnight
|
|
"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
|
|
target_layer = layer_name.lower()
|
|
if target_layer in ["gray", "white"]:
|
|
target_layer = "white"
|
|
|
|
map_name = f"vworld_{target_layer}.png"
|
|
map_path = proj_dir / map_name
|
|
|
|
if not map_path.exists() and target_layer == "satellite":
|
|
map_path = proj_dir / "vworld_map.png"
|
|
|
|
if not map_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"VWorld {layer_name} (매핑: {target_layer}) 지도가 존재하지 않습니다.")
|
|
return FileResponse(map_path, media_type="image/png")
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/geojson")
|
|
def get_project_geojson(project_id: str, layer: str) -> dict[str, Any]:
|
|
"""저장된 프로젝트의 특정 GeoJSON 레이어 데이터를 반환합니다.
|
|
용량이 매우 큰 '등고선' 레이어의 경우 백엔드에서 10배 이상 압축하는 단순화 처리를 적용하여 서빙합니다.
|
|
layer: '지적도', '용도지역', '행정구역_시군구', '행정구역_읍면동', '수계망', '등고선', '산사태'
|
|
"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
|
|
layer_mapping = {
|
|
"지적도": "연속지적도_bounds.geojson",
|
|
"용도지역": "용도지역도_bounds.geojson",
|
|
"행정구역_시군구": "행정구역_시군구_bounds.geojson",
|
|
"행정구역_읍면동": "행정구역_읍면동_bounds.geojson",
|
|
"수계망": "수계망_물줄기_bounds.geojson",
|
|
"등고선": "등고선_bounds.geojson",
|
|
"산사태": "산사태위험등급_bounds.geojson",
|
|
}
|
|
|
|
filename = layer_mapping.get(layer)
|
|
if not filename:
|
|
raise HTTPException(status_code=400, detail="유효하지 않은 레이어명입니다.")
|
|
|
|
filepath = proj_dir / filename
|
|
if not filepath.exists():
|
|
raise HTTPException(status_code=404, detail=f"요청한 레이어({layer}) 파일이 존재하지 않습니다.")
|
|
|
|
# 등고선 등 대용량 데이터 경량화 처리 (용량 24MB -> 2MB 내외로 압축)
|
|
if layer == "등고선":
|
|
simplified_filepath = proj_dir / "등고선_bounds_simplified.geojson"
|
|
# 캐싱된 단순화 파일이 존재하면 즉시 리턴
|
|
if simplified_filepath.exists():
|
|
try:
|
|
with open(simplified_filepath, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
|
|
# 단순화 파일 미존재 시: 실시간 경량화 연동 생성
|
|
try:
|
|
import geopandas as gpd
|
|
print(f"[{project_id}] 대용량 등고선 GeoJSON 단순화 압축 시작...")
|
|
gdf = gpd.read_file(filepath)
|
|
|
|
# 허용 거리 오차(tolerance) 설정 - 임도 설계 범위 30m 내외 정밀도 유지 수준 (경위도 도분초 0.00003도 기준)
|
|
# 형상은 깨지지 않고 복잡한 정밀 점 집합만 단순화
|
|
gdf['geometry'] = gdf['geometry'].simplify(tolerance=0.00003, preserve_topology=True)
|
|
|
|
# GeoJSON 파일로 백엔드 저장 캐싱
|
|
gdf.to_file(simplified_filepath, driver="GeoJSON")
|
|
print(f"[{project_id}] 등고선 단순화 완료! 용량 축소 성공.")
|
|
|
|
with open(simplified_filepath, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"[Warning] 등고선 단순화 처리 실패 (원본 전송): {e}")
|
|
|
|
try:
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"GeoJSON 파일 로드 실패: {e}")
|
|
|
|
|
|
@app.get("/tiles/{project_id}/{layer}/{z}/{x}/{y}.pbf")
|
|
def get_vector_tile(project_id: str, layer: str, z: int, x: int, y: int) -> Response:
|
|
"""프로젝트의 특정 레이어에 대한 정밀 벡터 타일(MVT) 조각을 동적으로 렌더링하여 반환합니다.
|
|
layer: '지적도', '용도지역', '행정구역_시군구', '행정구역_읍면동', '수계망', '등고선', '산사태', '임도노선'
|
|
"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
|
|
layer_mapping = {
|
|
"지적도": "연속지적도_bounds.geojson",
|
|
"용도지역": "용도지역도_bounds.geojson",
|
|
"행정구역_시군구": "행정구역_시군구_bounds.geojson",
|
|
"행정구역_읍면동": "행정구역_읍면동_bounds.geojson",
|
|
"수계망": "수계망_물줄기_bounds.geojson",
|
|
"등고선": "등고선_bounds.geojson",
|
|
"산사태": "산사태위험등급_bounds.geojson",
|
|
"임도노선": "임도노선.geojson", # 임도노선 레이어 지원 추가
|
|
}
|
|
|
|
filename = layer_mapping.get(layer)
|
|
if not filename:
|
|
raise HTTPException(status_code=400, detail="유효하지 않은 레이어명입니다.")
|
|
|
|
filepath = proj_dir / filename
|
|
|
|
# 임도 노선일 경우, EPSG:5176 -> EPSG:4326으로 변환된 파일이 없을 때 원본 SHP에서 실시간 재투영 생성
|
|
if layer == "임도노선" and not filepath.exists():
|
|
# samples 또는 output 디렉토리에서 원본 폴리라인 SHP을 검색 및 변환 처리
|
|
# T-102에서 구체화할 예정이나 API 구조 차원 선대응
|
|
road_shp_candidates = list(ROOT_DIR.glob("samples/**/*_Polyline.shp"))
|
|
if road_shp_candidates:
|
|
try:
|
|
import geopandas as gpd
|
|
gdf = gpd.read_file(road_shp_candidates[0])
|
|
gdf = gdf.to_crs(epsg=4326)
|
|
gdf.to_file(filepath, driver="GeoJSON")
|
|
except Exception as e:
|
|
print(f"[Warning] 임도 노선 SHP 재투영 실패: {e}")
|
|
|
|
if not filepath.exists():
|
|
# 등고선 등 원본 파일이 아예 없다면 빈 타일 리턴 (에러 방지)
|
|
import mapbox_vector_tile
|
|
empty_tile = mapbox_vector_tile.encode([])
|
|
return Response(content=empty_tile, media_type="application/x-protobuf")
|
|
|
|
try:
|
|
cache_key = f"{project_id}_{layer}"
|
|
# mvt_helper를 이용해 z, x, y에 대응되는 mvt 바이너리 반환
|
|
mvt_bytes = generate_mvt_tile(filepath, cache_key, z, x, y, layer_name=layer)
|
|
return Response(
|
|
content=mvt_bytes,
|
|
media_type="application/x-protobuf",
|
|
headers={"Content-Encoding": "identity"} # 성능 향상 캐시 바인딩 가능
|
|
)
|
|
except Exception as e:
|
|
print(f"[Error] MVT 타일 렌더링 실패: {e}")
|
|
# 오류 시에도 빈 타일 응답을 전송해 맵뷰가 멈추거나 중단되지 않게 방어
|
|
import mapbox_vector_tile
|
|
empty_tile = mapbox_vector_tile.encode([])
|
|
return Response(content=empty_tile, media_type="application/x-protobuf")
|
|
|
|
|
|
@app.get("/", response_model=None)
|
|
def root() -> FileResponse | dict[str, object]:
|
|
index_path = FRONTEND_DIST / "index.html"
|
|
if index_path.exists():
|
|
return FileResponse(index_path)
|
|
return {
|
|
"name": "Forest Road Phase 1 Ground Filtering Backend",
|
|
"status": "running",
|
|
"frontend_dev": "http://localhost:5173",
|
|
}
|
|
|
|
|
|
@app.post("/api/check-sample")
|
|
def check_sample(body: CheckSampleRequest) -> dict[str, object]:
|
|
# samples/step1_scan 경로가 실제로 존재하고, 사용자가 선택한 파일명과 매칭되는지 확인
|
|
if SAMPLE_DIR.exists():
|
|
sample_filenames = {f.name for f in SAMPLE_DIR.iterdir() if f.is_file()}
|
|
matched = all(fn in sample_filenames for fn in body.filenames)
|
|
if matched:
|
|
return {"matched": True, "project_id": "upload_sample"}
|
|
return {"matched": False, "project_id": None}
|
|
|
|
|
|
@app.post("/api/upload")
|
|
async def upload_files(
|
|
las_file: UploadFile = File(...),
|
|
prj_file: UploadFile | None = File(None),
|
|
tfw_file: UploadFile | None = File(None),
|
|
tif_file: UploadFile | None = File(None),
|
|
) -> dict[str, object]:
|
|
"""새로운 LAS 파일 업로드 및 전체 알고리즘 선행 연산"""
|
|
# 업로드하는 파일 중 LAS 파일명이 개발용 샘플 폴더에 있는 파일과 매치된다면 'upload_sample'로 고정
|
|
is_sample_las = False
|
|
if SAMPLE_DIR.exists():
|
|
sample_las_names = {f.name for f in SAMPLE_DIR.glob("*.las")} | {f.name for f in SAMPLE_DIR.glob("*.laz")}
|
|
if las_file.filename in sample_las_names:
|
|
is_sample_las = True
|
|
|
|
project_id = "upload_sample" if is_sample_las else f"upload_{int(time.time())}"
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
raw_dir = proj_dir / "raw"
|
|
raw_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
dest_las = raw_dir / las_file.filename
|
|
content = await las_file.read()
|
|
dest_las.write_bytes(content)
|
|
|
|
if prj_file:
|
|
dest_prj = raw_dir / prj_file.filename
|
|
prj_content = await prj_file.read()
|
|
dest_prj.write_bytes(prj_content)
|
|
# result.prj 형태로 복사
|
|
(proj_dir / "result.prj").write_bytes(prj_content)
|
|
|
|
if tfw_file:
|
|
dest_tfw = raw_dir / tfw_file.filename
|
|
tfw_content = await tfw_file.read()
|
|
dest_tfw.write_bytes(tfw_content)
|
|
|
|
if tif_file:
|
|
dest_tif = raw_dir / tif_file.filename
|
|
tif_content = await tif_file.read()
|
|
dest_tif.write_bytes(tif_content)
|
|
|
|
# 파이프라인 트리거 (이미 업로드 및 연산 완료된 'upload_sample'이라면 중복 실행 스킵)
|
|
npz_path = proj_dir / "structured.npz"
|
|
workflow_path = proj_dir / "workflow.json"
|
|
if not workflow_path.exists():
|
|
with open(workflow_path, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"current_stage": "scan",
|
|
"completed": [],
|
|
"stale_from": None,
|
|
"stage1_confirmed": None
|
|
}, f, ensure_ascii=False, indent=2)
|
|
|
|
if project_id == "upload_sample" and npz_path.exists():
|
|
# 기존 필터 캐시는 유지하고 신규 지표면 모델 캐시만 확인/생성
|
|
ensure_terrain_models(proj_dir)
|
|
return {"project_id": project_id, "status": "completed", "skipped": True}
|
|
|
|
try:
|
|
process_pipeline(project_id, dest_las)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"파이프라인 실행 실패: {str(e)}")
|
|
|
|
return {"project_id": project_id, "status": "completed"}
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/analyze-ground")
|
|
def run_ground_analysis(project_id: str) -> dict[str, object]:
|
|
"""기존의 지면 분석 트리거 API (샘플 데이터 초기 연산용 지원)"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
npz_path = proj_dir / "structured.npz"
|
|
|
|
# 아직 구조화 캐시가 없는 경우 실행
|
|
if not npz_path.exists():
|
|
if project_id == "upload_sample":
|
|
# samples 디렉토리에서 매칭되는 데이터를 instance/upload_sample/raw 로 복사하여 연산 시작
|
|
las_candidates = list(SAMPLE_DIR.glob("*.las")) + list(SAMPLE_DIR.glob("*.laz"))
|
|
if not las_candidates:
|
|
raise HTTPException(status_code=404, detail="샘플 폴더에 LAS 파일이 존재하지 않습니다.")
|
|
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
raw_dir = proj_dir / "raw"
|
|
raw_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 샘플 데이터들 복사
|
|
for sample_file in SAMPLE_DIR.iterdir():
|
|
if sample_file.is_file():
|
|
shutil.copy(sample_file, raw_dir / sample_file.name)
|
|
|
|
# result.prj 백업 복사
|
|
prj_src = SAMPLE_DIR / "result.prj"
|
|
if prj_src.exists():
|
|
shutil.copy(prj_src, proj_dir / "result.prj")
|
|
|
|
las_in_raw = raw_dir / las_candidates[0].name
|
|
process_pipeline("upload_sample", las_in_raw)
|
|
else:
|
|
raise HTTPException(status_code=404, detail="지정된 프로젝트를 찾을 수 없거나 데이터가 유실되었습니다.")
|
|
else:
|
|
# 기존 샘플 캐시에도 신규 15개 지표면 모델을 추가합니다.
|
|
try:
|
|
ensure_terrain_models(proj_dir)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"지표면 모델 생성 실패: {exc}")
|
|
|
|
workflow_path = proj_dir / "workflow.json"
|
|
if not workflow_path.exists():
|
|
with open(workflow_path, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"current_stage": "scan",
|
|
"completed": [],
|
|
"stale_from": None,
|
|
"stage1_confirmed": None
|
|
}, f, ensure_ascii=False, indent=2)
|
|
|
|
# 각 필터의 적용 포인트 카운트 요약 정보 반환
|
|
data = np.load(npz_path)
|
|
total_pts = int(data["total_points"][0])
|
|
|
|
# bounds 데이터를 프론트엔드가 기대하는 구조 형태로 추출
|
|
raw_bounds = data["bounds"]
|
|
bounds = {
|
|
"x": [float(raw_bounds[0][0]), float(raw_bounds[0][1])],
|
|
"y": [float(raw_bounds[1][0]), float(raw_bounds[1][1])],
|
|
"z": [float(raw_bounds[2][0]), float(raw_bounds[2][1])]
|
|
}
|
|
|
|
return {
|
|
"status": "done",
|
|
"total_points": total_pts,
|
|
"bounds": bounds, # bounds 데이터 주입!
|
|
"filters_status": {
|
|
"grid_min_z": int(np.sum(np.load(proj_dir / "mask_grid_min_z.npy"))),
|
|
"csf": int(np.sum(np.load(proj_dir / "mask_csf.npy"))),
|
|
"pmf": int(np.sum(np.load(proj_dir / "mask_pmf.npy"))),
|
|
"ransac": int(np.sum(np.load(proj_dir / "mask_ransac.npy"))),
|
|
}
|
|
}
|
|
|
|
|
|
def _load_terrain_manifest(project_id: str) -> tuple[Path, dict[str, Any]]:
|
|
terrain_dir = INSTANCE_DIR / project_id / "terrain_models"
|
|
manifest_path = terrain_dir / "manifest.json"
|
|
if not manifest_path.exists():
|
|
raise HTTPException(status_code=404, detail="지표면 모델 캐시가 없습니다. 분석을 먼저 실행하세요.")
|
|
try:
|
|
return terrain_dir, json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise HTTPException(status_code=500, detail=f"지표면 모델 Manifest를 읽지 못했습니다: {exc}")
|
|
|
|
|
|
def _terrain_method_entry(manifest: dict[str, Any], source_filter: str, method: str) -> dict[str, Any]:
|
|
allowed_filters = set(TERRAIN_MODEL_CONFIG["source_filters"])
|
|
if source_filter not in allowed_filters or method not in MODEL_METHODS:
|
|
raise HTTPException(status_code=400, detail="지원하지 않는 기준 필터 또는 지표면 모델입니다.")
|
|
try:
|
|
entry = manifest["source_filters"][source_filter]["methods"][method]
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="요청한 지표면 모델 결과가 없습니다.")
|
|
return entry
|
|
|
|
|
|
class WorkflowState(BaseModel):
|
|
current_stage: str
|
|
completed: list[str]
|
|
stale_from: str | None = None
|
|
stage1_confirmed: dict[str, Any] | None = None
|
|
|
|
@app.get("/api/projects/{project_id}/workflow")
|
|
def get_project_workflow(project_id: str) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
workflow_path = proj_dir / "workflow.json"
|
|
if not workflow_path.exists():
|
|
return {
|
|
"current_stage": "scan",
|
|
"completed": [],
|
|
"stale_from": None,
|
|
"stage1_confirmed": None
|
|
}
|
|
try:
|
|
with open(workflow_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"워크플로우 로드 실패: {e}")
|
|
|
|
def _validate_confirmed_smoothing(project_id: str, confirmed: dict[str, Any] | None) -> None:
|
|
"""Re-validate the confirmed model's smoothing on the server (계획서 9/I-403):
|
|
smoothing is only valid for TIN/DTM and requires the smoothed artifact to exist."""
|
|
if not confirmed:
|
|
return
|
|
method = confirmed.get("method")
|
|
smooth = bool(confirmed.get("smooth"))
|
|
if not smooth:
|
|
return
|
|
if method not in ("tin", "dtm"):
|
|
raise HTTPException(status_code=400, detail=f"'{method}' 모델은 스무딩을 지원하지 않습니다.")
|
|
filt = confirmed.get("source_filter", "csf")
|
|
smooth_npz = INSTANCE_DIR / project_id / "terrain_models" / f"{method}_{filt}_smooth.npz"
|
|
if not smooth_npz.exists():
|
|
raise HTTPException(status_code=400, detail="스무딩 결과가 아직 준비되지 않아 확정할 수 없습니다.")
|
|
|
|
|
|
@app.put("/api/projects/{project_id}/workflow")
|
|
def update_project_workflow(project_id: str, state: WorkflowState) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
workflow_path = proj_dir / "workflow.json"
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
_validate_confirmed_smoothing(project_id, state.stage1_confirmed) # I-403
|
|
|
|
# If the confirmed stage-1 model changed, the downstream route is stale (I-404).
|
|
prev_confirmed = _load_confirmed(workflow_path)
|
|
try:
|
|
data = state.model_dump()
|
|
if prev_confirmed != data.get("stage1_confirmed"):
|
|
data["stale_from"] = "route"
|
|
with open(workflow_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
return data
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"워크플로우 저장 실패: {e}")
|
|
|
|
def _require_finite(v: float | None, name: str) -> float | None:
|
|
"""Reject NaN/Infinity coordinate or numeric inputs (계획서 I-104/7.2)."""
|
|
if v is not None and not math.isfinite(v):
|
|
raise ValueError(f"{name} 값이 유효하지 않습니다(NaN/Infinity 불가).")
|
|
return v
|
|
|
|
|
|
class Point3D(BaseModel):
|
|
x: float
|
|
y: float
|
|
z: float
|
|
|
|
@field_validator("x", "y", "z")
|
|
@classmethod
|
|
def _finite(cls, v, info):
|
|
return _require_finite(v, info.field_name)
|
|
|
|
class CPPoint(BaseModel):
|
|
order: int
|
|
x: float
|
|
y: float
|
|
z: float
|
|
|
|
@field_validator("x", "y", "z")
|
|
@classmethod
|
|
def _finite(cls, v, info):
|
|
return _require_finite(v, info.field_name)
|
|
|
|
class APoint(BaseModel):
|
|
x: float
|
|
y: float
|
|
z: float | None = None
|
|
radius_m: float
|
|
|
|
@field_validator("x", "y", "z")
|
|
@classmethod
|
|
def _finite(cls, v, info):
|
|
return _require_finite(v, info.field_name)
|
|
|
|
@field_validator("radius_m")
|
|
@classmethod
|
|
def _radius_range(cls, v):
|
|
lo, hi = app_config.ROUTE_AVOID_RADIUS_RANGE_M
|
|
if not math.isfinite(v) or not (lo <= v <= hi):
|
|
raise ValueError(f"반경(radius_m)은 {lo}~{hi}m 범위여야 합니다.")
|
|
return v
|
|
|
|
class RoutePoints(BaseModel):
|
|
bp: Point3D | None = None
|
|
ep: Point3D | None = None
|
|
cp: list[CPPoint] = []
|
|
ap: list[APoint] = []
|
|
fp: list[APoint] = [] # 금지구역(FP): AP와 동일한 중심+반경 구조, 내부는 통행 불가
|
|
|
|
@app.get("/api/projects/{project_id}/route/points")
|
|
def get_route_points(project_id: str) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
route_dir = proj_dir / "route_design"
|
|
points_path = route_dir / "points.json"
|
|
if not points_path.exists():
|
|
return {"bp": None, "ep": None, "cp": [], "ap": [], "fp": []}
|
|
try:
|
|
with open(points_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
data.setdefault("fp", []) # 구버전 파일 호환
|
|
return data
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"포인트 로드 실패: {e}")
|
|
|
|
@app.put("/api/projects/{project_id}/route/points")
|
|
def save_route_points(project_id: str, points: RoutePoints) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
route_dir = proj_dir / "route_design"
|
|
route_dir.mkdir(parents=True, exist_ok=True)
|
|
points_path = route_dir / "points.json"
|
|
_validate_points_in_bounds(project_id, points) # 영역 밖 좌표 차단 (I-104)
|
|
_validate_points_not_in_fp(points) # 필수점이 FP 내부면 차단 (I-203)
|
|
try:
|
|
data = points.model_dump()
|
|
with open(points_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
_patch_workflow_stale(project_id, "route") # 포인트 변경 -> 경로 stale (I-404)
|
|
return data
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"포인트 저장 실패: {e}")
|
|
|
|
|
|
@app.delete("/api/projects/{project_id}/route/points")
|
|
def reset_route_points(project_id: str) -> dict[str, Any]:
|
|
"""Atomically clear all placed points and the route result (계획서 11.1/I-402).
|
|
Design options (options.json) are intentionally preserved."""
|
|
route_dir = INSTANCE_DIR / project_id / "route_design"
|
|
route_dir.mkdir(parents=True, exist_ok=True)
|
|
empty_points = {"bp": None, "ep": None, "cp": [], "ap": [], "fp": []}
|
|
try:
|
|
with open(route_dir / "points.json", "w", encoding="utf-8") as f:
|
|
json.dump(empty_points, f, ensure_ascii=False, indent=2)
|
|
# Remove the now-stale route result so the UI shows a clean slate.
|
|
result_path = route_dir / "route_result.json"
|
|
if result_path.exists():
|
|
result_path.unlink()
|
|
_patch_workflow_stale(project_id, "route") # 초기화 -> 경로 stale (I-404)
|
|
return {"status": "reset", "points": empty_points}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"배치 초기화 실패: {e}")
|
|
|
|
class RouteSolveRequest(BaseModel):
|
|
grade_class: str
|
|
min_curve_radius_m: float
|
|
paved: bool
|
|
# 경로 탐색 알고리즘: "dijkstra"(기존 비용면) | "ridge_valley"(능선-계곡 정속경사).
|
|
# 지형 보간법을 뜻하는 기존 method 와는 별개의 개념이다.
|
|
algorithm: str = "dijkstra"
|
|
# 가중치는 선택값: 미지정 시 solver 가 config.ROUTE_W_* 기본값을 사용한다.
|
|
weights: dict[str, float] | None = None
|
|
# 오르막/내리막 분리 한계(선택, 비율). 미지정 시 등급·포장 기반 max_grade 사용. (I-303)
|
|
max_uphill_grade: float | None = None
|
|
max_downhill_grade: float | None = None
|
|
# 오르막/내리막 분리 경사 하한(선택, 비율). ridge_valley 알고리즘 전용 —
|
|
# 상한(max_uphill_grade/max_downhill_grade)과 동일한 방식으로 방향별 적용한다.
|
|
# 미지정 시 config.ROUTE_ALT_MIN_GRADE(8%) 사용. 0 은 "하한 없음"을 뜻한다.
|
|
min_uphill_grade: float | None = None
|
|
min_downhill_grade: float | None = None
|
|
# AP 회피구역 우회 불가 시 통과 허용(고급). 기본 꺼짐. (I-202)
|
|
allow_avoid_pass_through: bool = False
|
|
|
|
@field_validator("algorithm")
|
|
@classmethod
|
|
def _algorithm_enum(cls, v):
|
|
if v not in app_config.ROUTE_ALGORITHMS:
|
|
raise ValueError(f"algorithm 은 {app_config.ROUTE_ALGORITHMS} 중 하나여야 합니다.")
|
|
return v
|
|
|
|
@field_validator("max_uphill_grade", "max_downhill_grade")
|
|
@classmethod
|
|
def _grade_limit_range(cls, v):
|
|
if v is None:
|
|
return v
|
|
lo, hi = app_config.ROUTE_GRADE_LIMIT_RANGE
|
|
if not math.isfinite(v) or not (lo < v <= hi):
|
|
raise ValueError(f"오르막/내리막 기울기 한계는 {lo} 초과 {hi} 이하(비율)여야 합니다.")
|
|
return v
|
|
|
|
@field_validator("min_uphill_grade", "min_downhill_grade")
|
|
@classmethod
|
|
def _min_grade_limit_range(cls, v):
|
|
if v is None:
|
|
return v
|
|
lo, hi = app_config.ROUTE_GRADE_LIMIT_RANGE
|
|
if not math.isfinite(v) or not (lo <= v <= hi):
|
|
raise ValueError(f"오르막/내리막 경사 하한은 {lo} 이상 {hi} 이하(비율)여야 합니다.")
|
|
return v
|
|
|
|
@field_validator("grade_class")
|
|
@classmethod
|
|
def _grade_class_enum(cls, v):
|
|
if v not in app_config.ROUTE_GRADE_CLASSES:
|
|
raise ValueError(f"grade_class 는 {app_config.ROUTE_GRADE_CLASSES} 중 하나여야 합니다.")
|
|
return v
|
|
|
|
@field_validator("min_curve_radius_m")
|
|
@classmethod
|
|
def _curve_radius_range(cls, v):
|
|
lo, hi = app_config.ROUTE_MIN_CURVE_RADIUS_RANGE_M
|
|
if not math.isfinite(v) or not (lo <= v <= hi):
|
|
raise ValueError(f"최소 곡선반지름은 {lo}~{hi}m 범위여야 합니다.")
|
|
return v
|
|
|
|
@field_validator("weights")
|
|
@classmethod
|
|
def _weights_valid(cls, v):
|
|
if v is None:
|
|
return v
|
|
for key, val in v.items():
|
|
if not math.isfinite(val) or val < 0 or val > app_config.ROUTE_WEIGHT_MAX:
|
|
raise ValueError(
|
|
f"가중치 '{key}' 는 0 이상 {app_config.ROUTE_WEIGHT_MAX} 이하의 유한값이어야 합니다."
|
|
)
|
|
return v
|
|
|
|
|
|
def _route_input_signature(
|
|
points_data: dict[str, Any],
|
|
confirmed: dict[str, Any] | None,
|
|
options_data: dict[str, Any] | None,
|
|
) -> str:
|
|
"""A stable hash of every input that affects a route result. Changing the placed
|
|
points, the confirmed stage-1 model, or the design options invalidates the saved
|
|
route (계획서 6.3: 포인트/확정옵션 변경 시 route_result 를 stale 로 표기)."""
|
|
payload = {
|
|
"points": points_data,
|
|
"confirmed": {
|
|
"source_filter": (confirmed or {}).get("source_filter"),
|
|
"method": (confirmed or {}).get("method"),
|
|
"smooth": (confirmed or {}).get("smooth"),
|
|
},
|
|
"options": options_data or {},
|
|
}
|
|
blob = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
|
|
return hashlib.sha256(blob).hexdigest()
|
|
|
|
|
|
def _load_confirmed(workflow_path: Path) -> dict[str, Any] | None:
|
|
if not workflow_path.exists():
|
|
return None
|
|
try:
|
|
with open(workflow_path, "r", encoding="utf-8") as f:
|
|
return (json.load(f) or {}).get("stage1_confirmed")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _patch_workflow_stale(project_id: str, stale_from: str | None) -> None:
|
|
"""Update only workflow.json's stale_from (계획서 10.1/I-404) without disturbing
|
|
the rest of the workflow state. Best-effort: never raises."""
|
|
wf_path = INSTANCE_DIR / project_id / "workflow.json"
|
|
if not wf_path.exists():
|
|
return
|
|
try:
|
|
with open(wf_path, "r", encoding="utf-8") as f:
|
|
wf = json.load(f) or {}
|
|
if wf.get("stale_from") != stale_from:
|
|
wf["stale_from"] = stale_from
|
|
with open(wf_path, "w", encoding="utf-8") as f:
|
|
json.dump(wf, f, ensure_ascii=False, indent=2)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _project_xy_bounds(project_id: str) -> tuple[float, float, float, float] | None:
|
|
"""Project (x_min, x_max, y_min, y_max) from structured.npz, or None if unknown."""
|
|
npz_path = INSTANCE_DIR / project_id / "structured.npz"
|
|
if not npz_path.exists():
|
|
return None
|
|
try:
|
|
data = np.load(npz_path)
|
|
b = data["bounds"]
|
|
return float(b[0][0]), float(b[0][1]), float(b[1][0]), float(b[1][1])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _validate_points_in_bounds(project_id: str, points: "RoutePoints") -> None:
|
|
"""Reject points outside the project area (계획서 I-104/7.2). A small margin
|
|
tolerates points placed right on the boundary."""
|
|
bounds = _project_xy_bounds(project_id)
|
|
if bounds is None:
|
|
return # bounds unknown -> skip (don't block saving)
|
|
x_min, x_max, y_min, y_max = bounds
|
|
margin = max((x_max - x_min), (y_max - y_min)) * 0.02 # 2% slack
|
|
|
|
def _check(label: str, x: float, y: float) -> None:
|
|
if not (x_min - margin <= x <= x_max + margin and y_min - margin <= y <= y_max + margin):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"{label} 좌표가 프로젝트 영역을 벗어났습니다 (x={x:.1f}, y={y:.1f}).",
|
|
)
|
|
|
|
if points.bp:
|
|
_check("BP", points.bp.x, points.bp.y)
|
|
if points.ep:
|
|
_check("EP", points.ep.x, points.ep.y)
|
|
for cp in points.cp:
|
|
_check(f"CP{cp.order}", cp.x, cp.y)
|
|
for i, ap in enumerate(points.ap):
|
|
_check(f"AP{i+1}", ap.x, ap.y)
|
|
for i, fp in enumerate(points.fp):
|
|
_check(f"FP{i+1}", fp.x, fp.y)
|
|
|
|
|
|
def _validate_points_not_in_fp(points: "RoutePoints") -> None:
|
|
"""Required points (BP/EP/CP) must not lie inside any forbidden zone (FP).
|
|
(계획서 3.3/I-203: BP/EP/CP가 FP 안에 놓이면 저장 또는 탐색 전에 오류)"""
|
|
if not points.fp:
|
|
return
|
|
|
|
def _inside_any(x: float, y: float):
|
|
for i, fp in enumerate(points.fp):
|
|
if math.hypot(x - fp.x, y - fp.y) < fp.radius_m:
|
|
return i + 1
|
|
return None
|
|
|
|
targets = []
|
|
if points.bp:
|
|
targets.append(("BP", points.bp.x, points.bp.y))
|
|
if points.ep:
|
|
targets.append(("EP", points.ep.x, points.ep.y))
|
|
targets += [(f"CP{cp.order}", cp.x, cp.y) for cp in points.cp]
|
|
for label, x, y in targets:
|
|
fp_no = _inside_any(x, y)
|
|
if fp_no is not None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"{label} 가 금지구역 FP{fp_no} 내부에 있습니다. 포인트 또는 FP를 옮기세요.",
|
|
)
|
|
|
|
@app.post("/api/projects/{project_id}/route/solve")
|
|
def run_route_solve(project_id: str, body: RouteSolveRequest) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
workflow_path = proj_dir / "workflow.json"
|
|
points_path = proj_dir / "route_design" / "points.json"
|
|
options_path = proj_dir / "route_design" / "options.json"
|
|
result_path = proj_dir / "route_design" / "route_result.json"
|
|
|
|
if not points_path.exists():
|
|
raise HTTPException(status_code=400, detail="배치된 포인트(BP/EP) 정보가 존재하지 않습니다.")
|
|
|
|
try:
|
|
with open(points_path, "r", encoding="utf-8") as f:
|
|
points_data = json.load(f)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"포인트 로드 실패: {e}")
|
|
|
|
# 경로 비용면은 1단계에서 사용자가 확정한 (필터 + 지표면 모델) 조합을 그대로 사용한다.
|
|
confirmed = _load_confirmed(workflow_path)
|
|
filter_key = (confirmed or {}).get("source_filter", "csf")
|
|
method = (confirmed or {}).get("method", "dtm")
|
|
smooth = (confirmed or {}).get("smooth", True)
|
|
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
(proj_dir / "route_design").mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
options_data = body.model_dump()
|
|
with open(options_path, "w", encoding="utf-8") as f:
|
|
json.dump(options_data, f, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"옵션 저장 실패: {e}")
|
|
|
|
# 알고리즘 선택: 기존 Dijkstra(기본) 또는 능선-계곡 정속경사 (계획서 3단계)
|
|
if body.algorithm == "ridge_valley":
|
|
from utils.route_solver_ridgevalley import solve_ridge_valley_route as _solver
|
|
else:
|
|
from utils.route_solver import solve_optimal_route as _solver
|
|
try:
|
|
result = _solver(
|
|
project_id=project_id,
|
|
filter_key=filter_key,
|
|
smooth=smooth,
|
|
points_data=points_data,
|
|
options=options_data,
|
|
instance_dir=INSTANCE_DIR,
|
|
method=method
|
|
)
|
|
|
|
# Stamp the inputs that produced this route so GET can later detect staleness.
|
|
result["input_signature"] = _route_input_signature(points_data, confirmed, options_data)
|
|
result["stale"] = False
|
|
|
|
with open(result_path, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
|
|
_patch_workflow_stale(project_id, None) # 재탐색 성공 -> stale 해제 (I-404)
|
|
return result
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"경로 탐색 실패: {e}")
|
|
|
|
@app.get("/api/projects/{project_id}/route/result")
|
|
def get_route_result(project_id: str) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
result_path = proj_dir / "route_design" / "route_result.json"
|
|
if not result_path.exists():
|
|
return {
|
|
"polyline": [],
|
|
"metrics": {
|
|
"length_m": 0.0,
|
|
"avg_grade_pct": 0.0,
|
|
"max_grade_pct": 0.0,
|
|
"slope_violations": 0
|
|
}
|
|
}
|
|
try:
|
|
with open(result_path, "r", encoding="utf-8") as f:
|
|
result = json.load(f)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"경로 결과 조회 실패: {e}")
|
|
|
|
# Recompute the current input signature; if it differs from the one the saved route
|
|
# was built with, the route is stale (points / confirmed model / options changed).
|
|
saved_sig = result.get("input_signature")
|
|
if saved_sig:
|
|
points_path = proj_dir / "route_design" / "points.json"
|
|
options_path = proj_dir / "route_design" / "options.json"
|
|
try:
|
|
points_data = json.loads(points_path.read_text(encoding="utf-8")) if points_path.exists() else {}
|
|
except Exception:
|
|
points_data = {}
|
|
try:
|
|
options_data = json.loads(options_path.read_text(encoding="utf-8")) if options_path.exists() else {}
|
|
except Exception:
|
|
options_data = {}
|
|
confirmed = _load_confirmed(proj_dir / "workflow.json")
|
|
current_sig = _route_input_signature(points_data, confirmed, options_data)
|
|
result["stale"] = (current_sig != saved_sig)
|
|
|
|
return result
|
|
|
|
|
|
class RouteConfirmRequest(BaseModel):
|
|
# User acknowledgement for soft AP intrusion warnings (I-405).
|
|
# Curve-radius shortfalls remain visible in metrics but never block confirmation.
|
|
acknowledge_warnings: bool = False
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/route/confirm")
|
|
def confirm_route(project_id: str, body: RouteConfirmRequest) -> dict[str, Any]:
|
|
"""Confirm the current route as the stage-2 baseline (계획서 10.2/I-405).
|
|
|
|
Hard gates: result exists, not stale, no FP intrusion, BP/CP/EP within tolerance,
|
|
no longitudinal-grade hard violation. AP intrusion requires explicit
|
|
acknowledgement; curve-radius shortfalls are informational only."""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
route_dir = proj_dir / "route_design"
|
|
result_path = route_dir / "route_result.json"
|
|
workflow_path = proj_dir / "workflow.json"
|
|
|
|
if not result_path.exists():
|
|
raise HTTPException(status_code=400, detail="확정할 경로 결과가 없습니다. 먼저 최적경로를 찾으세요.")
|
|
|
|
# Reuse GET's freshly-recomputed stale verdict + the persisted result.
|
|
result = get_route_result(project_id)
|
|
if not result.get("polyline"):
|
|
raise HTTPException(status_code=400, detail="경로 폴리라인이 비어 있어 확정할 수 없습니다.")
|
|
if result.get("stale"):
|
|
raise HTTPException(status_code=409, detail="입력이 변경되어 경로가 stale 상태입니다. 재탐색 후 확정하세요.")
|
|
|
|
# Hard gates.
|
|
if any(fi.get("intrudes") for fi in result.get("forbidden_intrusions", [])):
|
|
raise HTTPException(status_code=409, detail="경로가 금지구역(FP)을 통과합니다. 확정할 수 없습니다.")
|
|
if not result.get("required_points_ok", False):
|
|
raise HTTPException(status_code=409, detail="BP/CP/EP 통과 허용오차(1m)를 만족하지 않습니다.")
|
|
if result.get("metrics", {}).get("slope_violations", 0) > 0:
|
|
raise HTTPException(status_code=409, detail="종단경사 한계를 초과하는 구간이 있어 확정할 수 없습니다.")
|
|
|
|
# AP is a soft warning that needs acknowledgement. Curve-radius shortfalls are
|
|
# deliberately non-blocking: keep them in the result for review, but allow the
|
|
# user to confirm the route without an additional gate.
|
|
ap_warn = any(ai.get("intrudes") for ai in result.get("avoid_intrusions", []))
|
|
if ap_warn and not body.acknowledge_warnings:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="회피구역(AP) 경고가 있습니다. 확인 후 다시 확정하세요.",
|
|
)
|
|
|
|
# Stamp confirmation into route_result.json.
|
|
result["status"] = "confirmed"
|
|
result["confirmed_at"] = datetime.now(timezone.utc).isoformat()
|
|
result["confirmed_input_signature"] = result.get("input_signature")
|
|
try:
|
|
with open(result_path, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"경로 확정 저장 실패: {e}")
|
|
|
|
# Advance the workflow: route completed, stale cleared, ready for the section stage.
|
|
try:
|
|
wf = json.loads(workflow_path.read_text(encoding="utf-8")) if workflow_path.exists() else {}
|
|
except Exception:
|
|
wf = {}
|
|
completed = list(dict.fromkeys((wf.get("completed") or []) + ["scan", "route"]))
|
|
wf["completed"] = completed
|
|
# 경로 확정과 3단계 진입은 분리한다. 사용자가 별도의
|
|
# '종·횡단 생성 →' 버튼을 눌러 section 페이지로 이동한다.
|
|
wf["stale_from"] = "section" if (proj_dir / "sections" / "result.json").exists() else None
|
|
wf["current_stage"] = "route"
|
|
try:
|
|
with open(workflow_path, "w", encoding="utf-8") as f:
|
|
json.dump(wf, f, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"워크플로우 갱신 실패: {e}")
|
|
|
|
return {"status": "confirmed", "confirmed_at": result["confirmed_at"], "workflow": wf}
|
|
|
|
|
|
class SectionGenerateRequest(BaseModel):
|
|
station_interval_m: float = app_config.SECTION_STATION_INTERVAL_M
|
|
cross_half_width_m: float = app_config.SECTION_CROSS_HALF_WIDTH_M
|
|
cross_sample_interval_m: float = app_config.SECTION_CROSS_SAMPLE_INTERVAL_M
|
|
long_sample_interval_m: float = app_config.SECTION_LONG_SAMPLE_INTERVAL_M
|
|
vertical_exaggeration: float = app_config.SECTION_VERTICAL_EXAGGERATION
|
|
include_endpoint: bool = app_config.SECTION_INCLUDE_ENDPOINT
|
|
|
|
@field_validator(
|
|
"station_interval_m", "cross_half_width_m", "cross_sample_interval_m",
|
|
"long_sample_interval_m", "vertical_exaggeration",
|
|
)
|
|
@classmethod
|
|
def _positive_finite(cls, value, info):
|
|
if not math.isfinite(value) or value <= 0:
|
|
raise ValueError(f"{info.field_name}은 0보다 큰 유한한 값이어야 합니다.")
|
|
return value
|
|
|
|
|
|
def _atomic_write_json(path: Path, value: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def _section_input_signature(
|
|
route_result: dict[str, Any],
|
|
confirmed: dict[str, Any],
|
|
options: dict[str, Any],
|
|
terrain_dir: Path,
|
|
) -> str:
|
|
method = str(confirmed.get("method", "dtm"))
|
|
source_filter = str(confirmed.get("source_filter", "csf"))
|
|
smooth = bool(confirmed.get("smooth", False)) and method in {"dtm", "tin"}
|
|
suffix = "_smooth" if smooth else ""
|
|
model_path = terrain_dir / f"{method}_{source_filter}{suffix}.npz"
|
|
model_stat = None
|
|
if model_path.exists():
|
|
stat = model_path.stat()
|
|
model_stat = [model_path.name, stat.st_size, stat.st_mtime_ns]
|
|
payload = {
|
|
"schema_version": 1,
|
|
"route_signature": route_result.get("confirmed_input_signature"),
|
|
"confirmed": confirmed,
|
|
"options": options,
|
|
"model": model_stat,
|
|
}
|
|
return hashlib.sha256(
|
|
json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def _load_confirmed_route_for_sections(project_id: str) -> tuple[Path, dict[str, Any], dict[str, Any]]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
result = get_route_result(project_id)
|
|
if result.get("status") != "confirmed" or not result.get("confirmed_input_signature"):
|
|
raise HTTPException(status_code=409, detail="확정된 경로가 없습니다. 2단계에서 경로를 먼저 확정하세요.")
|
|
if result.get("stale"):
|
|
raise HTTPException(status_code=409, detail="확정 경로가 stale 상태입니다. 재탐색·재확정이 필요합니다.")
|
|
confirmed = _load_confirmed(proj_dir / "workflow.json")
|
|
if not confirmed:
|
|
raise HTTPException(status_code=409, detail="1단계 확정 지표면 정보가 없습니다.")
|
|
return proj_dir, result, confirmed
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/sections/generate")
|
|
def generate_project_sections(project_id: str, body: SectionGenerateRequest) -> dict[str, Any]:
|
|
build_key = str((INSTANCE_DIR / project_id / "sections").resolve())
|
|
with _ACTIVE_SECTION_BUILDS_GUARD:
|
|
if build_key in _ACTIVE_SECTION_BUILDS:
|
|
raise HTTPException(status_code=409, detail="종·횡단 생성이 이미 진행 중이어서 중복 요청을 취소했습니다.")
|
|
_ACTIVE_SECTION_BUILDS.add(build_key)
|
|
|
|
try:
|
|
proj_dir, route_result, confirmed = _load_confirmed_route_for_sections(project_id)
|
|
polyline = route_result.get("polyline") or []
|
|
if len(polyline) < 2:
|
|
raise HTTPException(status_code=400, detail="확정 경로 폴리라인이 비어 있습니다.")
|
|
|
|
options_data = body.model_dump()
|
|
estimated_length = float(route_result.get("metrics", {}).get("length_m", 0.0))
|
|
estimated_count = int(math.ceil(estimated_length / body.station_interval_m)) + 1
|
|
if estimated_count > app_config.SECTION_MAX_STATION_COUNT:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"예상 횡단도 {estimated_count:,}개가 최대 {app_config.SECTION_MAX_STATION_COUNT:,}개를 초과합니다.",
|
|
)
|
|
|
|
from utils.section_generator import SectionGenerationOptions, generate_sections
|
|
from utils.surface_elevation_sampler import build_surface_sampler
|
|
|
|
terrain_dir = proj_dir / "terrain_models"
|
|
sampler = build_surface_sampler(
|
|
terrain_dir,
|
|
str(confirmed.get("source_filter", "csf")),
|
|
str(confirmed.get("method", "dtm")),
|
|
bool(confirmed.get("smooth", False)),
|
|
)
|
|
generation_options = SectionGenerationOptions(
|
|
station_interval_m=body.station_interval_m,
|
|
cross_half_width_m=body.cross_half_width_m,
|
|
cross_sample_interval_m=body.cross_sample_interval_m,
|
|
long_sample_interval_m=body.long_sample_interval_m,
|
|
include_endpoint=body.include_endpoint,
|
|
)
|
|
prj_path = proj_dir / "result.prj"
|
|
crs = prj_path.read_text(encoding="utf-8", errors="replace") if prj_path.exists() else None
|
|
source_snapshot = {
|
|
"source_filter": confirmed.get("source_filter"),
|
|
"surface_method": confirmed.get("method"),
|
|
"smooth": bool(confirmed.get("smooth", False)),
|
|
"route_confirmed_signature": route_result.get("confirmed_input_signature"),
|
|
}
|
|
result = generate_sections(
|
|
polyline,
|
|
sampler,
|
|
generation_options,
|
|
source_snapshot=source_snapshot,
|
|
crs=crs,
|
|
)
|
|
result["options"]["vertical_exaggeration"] = body.vertical_exaggeration
|
|
result["input_signature"] = _section_input_signature(
|
|
route_result, confirmed, options_data, terrain_dir
|
|
)
|
|
result["generated_at"] = datetime.now(timezone.utc).isoformat()
|
|
result["stale"] = False
|
|
|
|
sections_dir = proj_dir / "sections"
|
|
_atomic_write_json(sections_dir / "options.json", options_data)
|
|
_atomic_write_json(sections_dir / "result.json", result)
|
|
|
|
workflow_path = proj_dir / "workflow.json"
|
|
try:
|
|
workflow = json.loads(workflow_path.read_text(encoding="utf-8")) if workflow_path.exists() else {}
|
|
workflow["stale_from"] = None
|
|
workflow["current_stage"] = "section"
|
|
_atomic_write_json(workflow_path, workflow)
|
|
except Exception:
|
|
pass
|
|
return result
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"종·횡단 생성 실패: {exc}")
|
|
finally:
|
|
with _ACTIVE_SECTION_BUILDS_GUARD:
|
|
_ACTIVE_SECTION_BUILDS.discard(build_key)
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/sections")
|
|
def get_project_sections(project_id: str) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
result_path = proj_dir / "sections" / "result.json"
|
|
if not result_path.exists():
|
|
raise HTTPException(status_code=404, detail="생성된 종·횡단 결과가 없습니다.")
|
|
try:
|
|
result = json.loads(result_path.read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"종·횡단 결과 로드 실패: {exc}")
|
|
|
|
try:
|
|
_, route_result, confirmed = _load_confirmed_route_for_sections(project_id)
|
|
options = result.get("options") or {}
|
|
current_signature = _section_input_signature(
|
|
route_result, confirmed, options, proj_dir / "terrain_models"
|
|
)
|
|
result["stale"] = current_signature != result.get("input_signature")
|
|
if result["stale"]:
|
|
result["status"] = "stale"
|
|
except HTTPException:
|
|
result["stale"] = True
|
|
result["status"] = "stale"
|
|
return result
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/sections/confirm")
|
|
def confirm_project_sections(project_id: str) -> dict[str, Any]:
|
|
result = get_project_sections(project_id)
|
|
if result.get("stale"):
|
|
raise HTTPException(status_code=409, detail="종·횡단 결과가 stale 상태입니다. 재생성 후 확정하세요.")
|
|
result["status"] = "confirmed"
|
|
result["confirmed_at"] = datetime.now(timezone.utc).isoformat()
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
_atomic_write_json(proj_dir / "sections" / "result.json", result)
|
|
|
|
workflow_path = proj_dir / "workflow.json"
|
|
try:
|
|
workflow = json.loads(workflow_path.read_text(encoding="utf-8")) if workflow_path.exists() else {}
|
|
workflow["completed"] = list(dict.fromkeys((workflow.get("completed") or []) + ["scan", "route", "section"]))
|
|
workflow["current_stage"] = "section"
|
|
workflow["stale_from"] = None
|
|
_atomic_write_json(workflow_path, workflow)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"종·횡단 workflow 확정 실패: {exc}")
|
|
result["workflow"] = workflow
|
|
return result
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/terrain-models")
|
|
def get_terrain_models(project_id: str) -> dict[str, Any]:
|
|
_, manifest = _load_terrain_manifest(project_id)
|
|
return manifest
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/terrain-models/{source_filter}/{method}/metadata")
|
|
def get_terrain_model_metadata(project_id: str, source_filter: str, method: str) -> dict[str, Any]:
|
|
_, manifest = _load_terrain_manifest(project_id)
|
|
return _terrain_method_entry(manifest, source_filter, method)
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/terrain-models/{source_filter}/{method}/preview")
|
|
def get_terrain_model_preview(
|
|
project_id: str,
|
|
source_filter: str,
|
|
method: str,
|
|
smooth: bool = False
|
|
) -> FileResponse:
|
|
terrain_dir, manifest = _load_terrain_manifest(project_id)
|
|
entry = _terrain_method_entry(manifest, source_filter, method)
|
|
|
|
# 스무딩본 요청 처리
|
|
if smooth:
|
|
smooth_entry = entry.get("smooth")
|
|
if not smooth_entry or smooth_entry.get("status") != "completed":
|
|
raise HTTPException(status_code=404, detail="스무딩 모델 계산이 완료되지 않았거나 지원하지 않습니다.")
|
|
preview_file = smooth_entry.get("preview_file")
|
|
media_type = smooth_entry.get("preview_media_type", "application/octet-stream")
|
|
preview_path = terrain_dir / Path(preview_file).name
|
|
else:
|
|
if entry.get("status") != "completed" or not entry.get("preview_file"):
|
|
raise HTTPException(status_code=409, detail=entry.get("error") or "지표면 모델 계산이 완료되지 않았습니다.")
|
|
preview_path = terrain_dir / Path(entry["preview_file"]).name
|
|
media_type = entry.get("preview_media_type", "application/octet-stream")
|
|
|
|
if not preview_path.exists():
|
|
raise HTTPException(status_code=404, detail="지표면 모델 미리보기 파일이 없습니다.")
|
|
return FileResponse(preview_path, media_type=media_type)
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/terrain-models/{source_filter}/{method}/contour")
|
|
def get_terrain_model_contour(
|
|
project_id: str,
|
|
source_filter: str,
|
|
method: str,
|
|
interval: float | None = None,
|
|
smooth: bool = False
|
|
) -> dict[str, Any]:
|
|
from utils.contour_extractor import CONTOUR_EXTRACTOR_VERSION, extract_contours
|
|
import config
|
|
|
|
terrain_dir, manifest = _load_terrain_manifest(project_id)
|
|
entry = _terrain_method_entry(manifest, source_filter, method)
|
|
|
|
# 1. 스무딩 설정에 따라 타겟 파일 및 정보 전환
|
|
if smooth:
|
|
smooth_entry = entry.get("smooth")
|
|
if not smooth_entry or smooth_entry.get("status") != "completed" or not smooth_entry.get("model_file"):
|
|
raise HTTPException(status_code=409, detail="스무딩 모델 계산이 완료되지 않았습니다.")
|
|
model_file = smooth_entry["model_file"]
|
|
representation = "regular_grid" if method == "dtm" else "triangular_mesh"
|
|
else:
|
|
if entry.get("status") != "completed" or not entry.get("model_file"):
|
|
raise HTTPException(status_code=409, detail="지표면 모델 계산이 완료되지 않았습니다.")
|
|
model_file = entry["model_file"]
|
|
representation = entry.get("representation", "regular_grid")
|
|
|
|
model_path = terrain_dir / Path(model_file).name
|
|
if not model_path.exists():
|
|
raise HTTPException(status_code=404, detail="지표면 모델 데이터 파일(npz)이 없습니다.")
|
|
|
|
# 2. 간격 결정 및 클램프
|
|
if interval is None:
|
|
interval = getattr(config, "CONTOUR_INTERVAL_METERS", 5.0)
|
|
|
|
min_interval = getattr(config, "CONTOUR_MIN_INTERVAL_METERS", 0.5)
|
|
interval = max(interval, min_interval)
|
|
|
|
# 3. 캐시 조회
|
|
suffix = "_smooth" if smooth else ""
|
|
cache_filename = f"contour_{source_filter}_{method}{suffix}_{interval}m.json"
|
|
cache_path = terrain_dir / cache_filename
|
|
|
|
if cache_path.exists():
|
|
try:
|
|
cached = json.loads(cache_path.read_text(encoding="utf-8"))
|
|
if cached.get("extractor_version") == CONTOUR_EXTRACTOR_VERSION:
|
|
return cached
|
|
except Exception:
|
|
pass # 손상 시 재추출
|
|
|
|
# 4. 추출 진행
|
|
try:
|
|
target_grid_m = getattr(config, "CONTOUR_GRID_RESOLUTION_METERS", 1.0)
|
|
bounds_info = manifest.get("bounds", {})
|
|
|
|
contours = extract_contours(
|
|
model_path,
|
|
representation=representation,
|
|
interval=interval,
|
|
target_grid_m=target_grid_m,
|
|
scene_center=None
|
|
)
|
|
|
|
response_data = {
|
|
"extractor_version": CONTOUR_EXTRACTOR_VERSION,
|
|
"project_id": project_id,
|
|
"source_filter": source_filter,
|
|
"method": method,
|
|
"interval": interval,
|
|
"bounds": bounds_info,
|
|
"contours": contours
|
|
}
|
|
|
|
# 캐시 저장
|
|
try:
|
|
cache_path.write_text(json.dumps(response_data, ensure_ascii=False), encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
return response_data
|
|
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"등고선 추출 실패: {exc}")
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/terrain-models/rebuild")
|
|
def rebuild_terrain_models(project_id: str) -> dict[str, Any]:
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
if not proj_dir.exists():
|
|
raise HTTPException(status_code=404, detail="프로젝트가 없습니다.")
|
|
try:
|
|
return ensure_terrain_models(proj_dir, force=True)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"지표면 모델 재생성 실패: {exc}")
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/points")
|
|
def get_filtered_points(project_id: str, method: str = "raw") -> dict[str, Any]:
|
|
"""사용자가 요청한 특정 필터(method)의 연산 결과 지표면 데이터를 반환.
|
|
"""
|
|
proj_dir = INSTANCE_DIR / project_id
|
|
npz_path = proj_dir / "structured.npz"
|
|
if not npz_path.exists():
|
|
raise HTTPException(status_code=404, detail="구조화된 포인트클라우드 데이터가 없습니다.")
|
|
|
|
data = np.load(npz_path)
|
|
xyz = data["xyz"]
|
|
rgb = data["rgb"]
|
|
|
|
# 필터 마스크 로드 및 마스킹
|
|
if method == "raw":
|
|
mask = np.ones(len(xyz), dtype=bool)
|
|
else:
|
|
mask_path = proj_dir / f"mask_{method}.npy"
|
|
if not mask_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"요청한 필터({method}) 결과가 없습니다.")
|
|
mask = np.load(mask_path)
|
|
|
|
filtered_xyz = xyz[mask]
|
|
filtered_rgb = rgb[mask]
|
|
|
|
# 클라이언트 부하 완화를 위해 포인트 다운샘플링 (config의 MAX_RENDER_POINTS 기준 적용)
|
|
max_display_pts = MAX_RENDER_POINTS
|
|
n_pts = len(filtered_xyz)
|
|
if n_pts > max_display_pts:
|
|
step = n_pts // max_display_pts
|
|
filtered_xyz = filtered_xyz[::step][:max_display_pts]
|
|
filtered_rgb = filtered_rgb[::step][:max_display_pts]
|
|
|
|
# center_offset이 npz에 없는 경우 bounds 정보로부터 중심점을 산출합니다.
|
|
try:
|
|
raw_offset = data["center_offset"]
|
|
center_offset = {
|
|
"x": float(raw_offset[0]),
|
|
"y": float(raw_offset[1]),
|
|
"z": float(raw_offset[2])
|
|
}
|
|
except KeyError:
|
|
bounds = data["bounds"]
|
|
center_offset = {
|
|
"x": float((bounds[0][0] + bounds[0][1]) / 2.0),
|
|
"y": float((bounds[1][0] + bounds[1][1]) / 2.0),
|
|
"z": float((bounds[2][0] + bounds[2][1]) / 2.0)
|
|
}
|
|
|
|
# bounds 데이터를 프론트엔드가 기대하는 구조 형태로 추출
|
|
raw_bounds = data["bounds"]
|
|
bounds = {
|
|
"x": [float(raw_bounds[0][0]), float(raw_bounds[0][1])],
|
|
"y": [float(raw_bounds[1][0]), float(raw_bounds[1][1])],
|
|
"z": [float(raw_bounds[2][0]), float(raw_bounds[2][1])]
|
|
}
|
|
|
|
# 셰이프 JSON 반환 규격 포맷팅
|
|
# 프론트엔드의 pointCloudUtils.ts 설계 규격에 맞추어 points와 rgb를 분리한 2D 배열 리스트로 반환합니다.
|
|
xyz_list = [[float(x), float(y), float(z)] for x, y, z in filtered_xyz]
|
|
rgb_list = [[int(r), int(g), int(b)] for r, g, b in filtered_rgb]
|
|
|
|
return {
|
|
"project_id": project_id,
|
|
"method": method,
|
|
"total_points": n_pts,
|
|
"points": xyz_list,
|
|
"rgb": rgb_list,
|
|
"center_offset": center_offset,
|
|
"bounds": bounds,
|
|
"viewer_config": get_point_cloud_viewer_config(),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
# py.exe 직접 실행 시 uvicorn 리로더가 모듈 위치를 찾을 수 있도록 시스템 경로 및 디렉토리 고정
|
|
current_dir = str(ROOT_DIR)
|
|
if current_dir not in sys.path:
|
|
sys.path.insert(0, current_dir)
|
|
|
|
# 백엔드 기동 시 프론트엔드 개발 서버를 자동으로 같이 띄웁니다.
|
|
frontend_path = ROOT_DIR / "frontend"
|
|
if frontend_path.exists():
|
|
print("[System] 백그라운드에서 프론트엔드 개발 서버(npm run dev)를 시작합니다...")
|
|
try:
|
|
# Windows OS에서 백그라운드로 띄우기 위해 Popen 사용 (shell=True)
|
|
subprocess.Popen(
|
|
"npm run dev",
|
|
shell=True,
|
|
cwd=str(frontend_path),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
except Exception as e:
|
|
print(f"[Warning] 프론트엔드 개발 서버 자동 실행 실패: {e}")
|
|
|
|
# reload_dirs를 명시하여 main.py를 포함한 백엔드 폴더 내부 변경을 감시
|
|
uvicorn.run(
|
|
"main:app",
|
|
host="0.0.0.0",
|
|
port=8000,
|
|
reload=True,
|
|
reload_dirs=[current_dir]
|
|
)
|