feat(B05): 셀별 흐름 방향 판정 + 화살표/적청 표기 (확장 미포함)
최외곽 셀부터 물길을 따라가며 도로 도달 여부를 판정하고 화면에 셀 단위로 표기한다. 격자 확장은 다음 검증 단계라 넣지 않았다. - classify_flow: 최외곽 셀에서 출발해 D8 수신 셀을 한 칸씩 따라가고, 경로가 끝나면 결과를 경로 전체에 되돌려 적는다. 한 번 판정한 셀은 재분석하지 않고 다른 경로가 만나면 즉시 결론을 가져온다. 최외곽으로 안 닿은 내부 셀은 그다음에 따로 출발시킨다. 합성 검증 165,874셀 0.5s. - outermost_cells: 해석 영역 밖에 8이웃이 닿는 셀 = 최외곽. - build_terrain_grid(domain=): 1차 영역 셀 마스크 안쪽만 해석 대상으로 삼는다. - direction_codes: 수신 셀 인덱스를 3x3 방향 코드로. 화살표 렌더용. - 응답은 셀당 1바이트 base64 (하위4비트=방향, 15=표고없음, 0x80=도로도달). 197,623셀 -> 257KB. - 프론트: 셀마다 화살표. 도로 도달 적색, 미도달 파랑, 표고없음 회색. 셀이 7px 미만이면 채움색만 남긴다. 격자선을 흰색으로 변경. - 저장: 02_flow_direction.npz (direction/reaches_road/analyzed) + manifest 합성 검증(능선 y=200, 도로 y=500): 도달 셀 y 200~502, 미도달 y 150~600. 능선 남쪽과 도로 아래가 정확히 제외됨. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -327,6 +327,20 @@ export interface DrainagePrimaryRegion {
|
||||
/** 실제 생성된 셀 구간 [행, 시작열, 끝열(포함)]. 낱개 셀 대신 구간으로 온다. */
|
||||
row_spans: Array<[number, number, number]>;
|
||||
};
|
||||
/** 셀별 흐름 방향과 도로 도달 여부. 등고선이 없어 판정을 못하면 null. */
|
||||
flow: {
|
||||
encoding: "base64-uint8";
|
||||
cells: number;
|
||||
reaches_road: number;
|
||||
no_road: number;
|
||||
/** 격자에는 있으나 등고선 TIN 밖이라 표고가 없어 판정 못한 셀. */
|
||||
unanalyzed: number;
|
||||
outer_seeds: number;
|
||||
interior_seeds: number;
|
||||
/** 셀당 1바이트. 하위 4비트=3×3 방향코드(4=제자리, 15=무효), 0x80=도로 도달.
|
||||
* 순서는 grid.row_spans를 행 → 구간 → 열 오름차순으로 훑은 순서와 같다. */
|
||||
data: string;
|
||||
} | null;
|
||||
/** 영구저장소에 남긴 검증용 GeoJSON 경로. */
|
||||
saved_to: string | null;
|
||||
}
|
||||
|
||||
@@ -36,7 +36,10 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
|
||||
is_uphill_at,
|
||||
)
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Flow import (
|
||||
FlowClassification,
|
||||
RoadRaster,
|
||||
border_contact,
|
||||
classify_flow,
|
||||
largest_ring,
|
||||
outer_boundary,
|
||||
polygonize_labels,
|
||||
@@ -45,6 +48,7 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Flow import (
|
||||
)
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
|
||||
GridSpec,
|
||||
TerrainGrid,
|
||||
build_contour_cloud,
|
||||
build_terrain_grid,
|
||||
expand_grid_spec,
|
||||
@@ -55,6 +59,7 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Stream import (
|
||||
build_primary_region,
|
||||
)
|
||||
from config.config_system import (
|
||||
DRAINAGE_CONTOUR_CLIP_MARGIN_M,
|
||||
DRAINAGE_DITCH_SAMPLE_M,
|
||||
DRAINAGE_EXPAND_STEP_M,
|
||||
DRAINAGE_GRID_SIZE_M,
|
||||
@@ -203,6 +208,57 @@ def preview_primary_region(
|
||||
return resolve_primary_region(vertices, route_line, contour_features, stream_features)
|
||||
|
||||
|
||||
@dataclass
|
||||
class StagePreview:
|
||||
"""단계 검증 산출물 묶음. 기능을 붙일 때마다 여기에 항목이 하나씩 늘어난다."""
|
||||
|
||||
region: PrimaryRegion
|
||||
terrain: TerrainGrid | None = None
|
||||
road: RoadRaster | None = None
|
||||
flow: FlowClassification | None = None
|
||||
|
||||
|
||||
def preview_stages(
|
||||
vertices: list[RouteVertex],
|
||||
contour_features: list[dict[str, Any]],
|
||||
stream_features: list[dict[str, Any]],
|
||||
) -> StagePreview | None:
|
||||
"""지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다.
|
||||
|
||||
현재 포함: ① 1차 배수유역 ② 격자 생성 ③ 표고·D8 ④ 흐름 방향/도로 도달 판정.
|
||||
**격자 확장은 넣지 않는다** — 다음 검증 단계다(2026-07-31 사용자 지시).
|
||||
"""
|
||||
if len(vertices) < 2:
|
||||
return None
|
||||
route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
|
||||
region = resolve_primary_region(vertices, route_line, contour_features, stream_features)
|
||||
if region is None:
|
||||
return None
|
||||
|
||||
# TIN은 격자 범위 + 여유만큼만 읽는다. 확장이 없으므로 여유는 클리핑 마진이면 충분하다.
|
||||
spec = region.spec
|
||||
cloud = build_contour_cloud(
|
||||
contour_features,
|
||||
route_elevation_floor([vertex.z for vertex in vertices]),
|
||||
(
|
||||
spec.x_min - DRAINAGE_CONTOUR_CLIP_MARGIN_M,
|
||||
spec.y_max - spec.n_rows * spec.cell_m - DRAINAGE_CONTOUR_CLIP_MARGIN_M,
|
||||
spec.x_min + spec.n_cols * spec.cell_m + DRAINAGE_CONTOUR_CLIP_MARGIN_M,
|
||||
spec.y_max + DRAINAGE_CONTOUR_CLIP_MARGIN_M,
|
||||
),
|
||||
)
|
||||
if cloud.is_empty:
|
||||
logger.warning("배수유역: 격자 범위 안에 등고선이 없어 흐름 판정을 건너뜁니다.")
|
||||
return StagePreview(region=region)
|
||||
|
||||
started = time.perf_counter()
|
||||
terrain = build_terrain_grid(spec, cloud, region.cell_mask)
|
||||
road = rasterize_road(spec, route_line, DRAINAGE_ROAD_WIDTH_M)
|
||||
flow = classify_flow(terrain, road)
|
||||
logger.info("배수유역: 흐름 판정 %.1fs (셀 %d개)", time.perf_counter() - started, spec.size)
|
||||
return StagePreview(region=region, terrain=terrain, road=road, flow=flow)
|
||||
|
||||
|
||||
# ── ③~④ 격자 해석 (캐시 대상) ───────────────────────────────────────────────
|
||||
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ logger = logging.getLogger(__name__)
|
||||
# 단계 이름 → 파일 접두 번호. 순서대로 읽으면 파이프라인 진행 순서가 된다.
|
||||
STAGES: dict[str, str] = {
|
||||
"primary_region": "01",
|
||||
"flow_direction": "02",
|
||||
}
|
||||
|
||||
_MANIFEST_FILENAME = "manifest.json"
|
||||
@@ -121,17 +122,19 @@ def _to_lonlat_coords(geometry: BaseGeometry, to_lonlat: LonLat) -> Any:
|
||||
return None
|
||||
|
||||
|
||||
def write_cell_mask(stored_path: str, stage: str, spec: Any, mask: Any) -> str | None:
|
||||
"""격자 셀 마스크를 `.npz`로 남긴다.
|
||||
def write_grid_arrays(
|
||||
stored_path: str, stage: str, spec: Any, arrays: dict[str, Any], summary: dict[str, Any]
|
||||
) -> str | None:
|
||||
"""격자 크기의 배열들을 `.npz`로 남긴다(셀 마스크·흐름 방향·도달 여부 등).
|
||||
|
||||
셀이 수십만 개라 GeoJSON 폴리곤으로는 못 남긴다. 격자 원점·셀 크기와 bool 마스크만
|
||||
저장하면 어느 셀이 생성됐는지 그대로 복원된다.
|
||||
셀이 수십만 개라 GeoJSON 폴리곤으로는 못 남긴다. 격자 원점·셀 크기와 배열만 저장하면
|
||||
어느 셀이 어떤 값이었는지 그대로 복원된다. 요약값은 manifest에도 기록한다.
|
||||
"""
|
||||
prefix = STAGES.get(stage)
|
||||
if prefix is None or mask is None:
|
||||
if prefix is None or not arrays:
|
||||
return None
|
||||
directory = drainage_dir(stored_path)
|
||||
target = directory / f"{prefix}_{stage}_cells.npz"
|
||||
target = directory / f"{prefix}_{stage}.npz"
|
||||
try:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
np.savez_compressed(
|
||||
@@ -141,12 +144,13 @@ def write_cell_mask(stored_path: str, stage: str, spec: Any, mask: Any) -> str |
|
||||
cell_m=spec.cell_m,
|
||||
n_rows=spec.n_rows,
|
||||
n_cols=spec.n_cols,
|
||||
mask=mask,
|
||||
**arrays,
|
||||
)
|
||||
except OSError:
|
||||
logger.warning("배수유역: 셀 마스크 저장 실패 (%s)", target)
|
||||
logger.warning("배수유역: %s 배열 저장 실패 (%s)", stage, target)
|
||||
return None
|
||||
logger.info("배수유역: 셀 마스크 저장 — %s (%d셀)", target, int(mask.sum()))
|
||||
_update_manifest(directory, f"{stage}_arrays", target.name, summary)
|
||||
logger.info("배수유역: %s 배열 저장 — %s (%s)", stage, target, ", ".join(arrays))
|
||||
return str(target)
|
||||
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ from shapely.ops import unary_union
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
|
||||
GridSpec,
|
||||
TerrainGrid,
|
||||
direction_codes,
|
||||
grid_transform,
|
||||
)
|
||||
from config.config_system import (
|
||||
@@ -39,6 +40,18 @@ logger = logging.getLogger(__name__)
|
||||
# 포인터 더블링 반복 상한. 한 번에 경로 길이가 2배가 되므로 2^40 스텝이면 어떤 격자도 덮는다.
|
||||
_MAX_DOUBLING_ROUNDS = 40
|
||||
|
||||
# 8이웃 (행 증분, 열 증분) — 최외곽 판정용. 거리는 쓰지 않는다.
|
||||
_NEIGHBOR_SHIFTS = (
|
||||
(-1, 0, 1.0),
|
||||
(1, 0, 1.0),
|
||||
(0, -1, 1.0),
|
||||
(0, 1, 1.0),
|
||||
(-1, -1, 1.0),
|
||||
(-1, 1, 1.0),
|
||||
(1, -1, 1.0),
|
||||
(1, 1, 1.0),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RoadRaster:
|
||||
@@ -158,6 +171,115 @@ def trace_flow(terrain: TerrainGrid, road: RoadRaster) -> FlowResult:
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FlowClassification:
|
||||
"""셀별 흐름 방향과 도로 도달 여부 — 확장 없이 현재 격자만 본 결과."""
|
||||
|
||||
direction: np.ndarray # (R*C,) int8 — 3×3 방향 코드(4=제자리), 무효 셀은 −1
|
||||
reaches_road: np.ndarray # (R*C,) bool — 물길을 따라가면 도로에 닿는가
|
||||
analyzed: np.ndarray # (R*C,) bool — 실제로 판정한 셀
|
||||
outer_seeds: int # 최외곽에서 출발해 판정한 셀 수
|
||||
interior_seeds: int # 최외곽 추적에 안 걸려 따로 출발시킨 내부 셀 수
|
||||
|
||||
|
||||
def outermost_cells(domain: np.ndarray) -> np.ndarray:
|
||||
"""해석 영역의 최외곽 셀 — 영역 밖(또는 격자 밖)에 8이웃이 하나라도 닿는 셀."""
|
||||
padded = np.zeros((domain.shape[0] + 2, domain.shape[1] + 2), dtype=bool)
|
||||
padded[1:-1, 1:-1] = domain
|
||||
exposed = np.zeros_like(domain)
|
||||
rows, cols = domain.shape
|
||||
for row_shift, col_shift, _ in _NEIGHBOR_SHIFTS:
|
||||
neighbour = padded[
|
||||
1 + row_shift : 1 + row_shift + rows, 1 + col_shift : 1 + col_shift + cols
|
||||
]
|
||||
exposed |= ~neighbour
|
||||
return exposed & domain
|
||||
|
||||
|
||||
def classify_flow(terrain: TerrainGrid, road: RoadRaster) -> FlowClassification:
|
||||
"""최외곽 셀부터 물길을 따라가며 도로 도달 여부를 판정한다.
|
||||
|
||||
최외곽 셀에서 출발해 D8 수신 셀을 한 칸씩 따라가고, 경로가 끝나면 그 결과를 경로 전체에
|
||||
되돌려 적는다. **한 번 판정한 셀은 다시 분석하지 않는다** — 다른 경로가 그 셀을 만나면
|
||||
거기서 즉시 결론을 가져온다. 최외곽 추적으로 안 닿은 내부 셀은 그다음에 따로 출발시킨다
|
||||
(2026-07-31 사용자 지시).
|
||||
|
||||
경로가 도로 셀에 닿으면 경로 전체가 도로 도달(적색), 싱크에서 멈추거나 해석 영역 밖으로
|
||||
나가면 미도달(파랑)이다. 격자 확장은 하지 않는다 — 다음 단계다.
|
||||
"""
|
||||
spec = terrain.spec
|
||||
valid = terrain.valid.reshape(-1)
|
||||
receiver = terrain.receiver
|
||||
is_road = road.mask.reshape(-1) & valid
|
||||
|
||||
direction = np.where(valid, direction_codes(spec, receiver), -1).astype(np.int8)
|
||||
reaches = np.zeros(spec.size, dtype=bool)
|
||||
# 0=미방문, 1=경로에 올라 있음, 2=판정 완료
|
||||
state = np.zeros(spec.size, dtype=np.int8)
|
||||
|
||||
outer = np.flatnonzero(outermost_cells(terrain.valid))
|
||||
remaining = np.flatnonzero(valid)
|
||||
outer_done = _walk_from(outer, receiver, valid, is_road, state, reaches)
|
||||
interior_done = _walk_from(remaining, receiver, valid, is_road, state, reaches)
|
||||
|
||||
analyzed = state == 2
|
||||
logger.info(
|
||||
"배수유역: 흐름 판정 %d셀 (최외곽 출발 %d / 내부 보충 %d) — 도로 도달 %d, 미도달 %d",
|
||||
int(analyzed.sum()),
|
||||
outer_done,
|
||||
interior_done,
|
||||
int((reaches & analyzed).sum()),
|
||||
int((~reaches & analyzed).sum()),
|
||||
)
|
||||
return FlowClassification(
|
||||
direction=direction,
|
||||
reaches_road=reaches,
|
||||
analyzed=analyzed,
|
||||
outer_seeds=outer_done,
|
||||
interior_seeds=interior_done,
|
||||
)
|
||||
|
||||
|
||||
def _walk_from(
|
||||
starts: np.ndarray,
|
||||
receiver: np.ndarray,
|
||||
valid: np.ndarray,
|
||||
is_road: np.ndarray,
|
||||
state: np.ndarray,
|
||||
reaches: np.ndarray,
|
||||
) -> int:
|
||||
"""출발 셀 목록에서 물길을 따라가며 판정한다. 새로 판정한 셀 수를 돌려준다."""
|
||||
resolved = 0
|
||||
path: list[int] = []
|
||||
for start in starts.tolist():
|
||||
if state[start] == 2:
|
||||
continue
|
||||
path.clear()
|
||||
node = start
|
||||
while True:
|
||||
if state[node] == 2:
|
||||
verdict = bool(reaches[node])
|
||||
break
|
||||
if state[node] == 1: # 방어: 채움·평탄해소 후에는 순환이 없어야 한다
|
||||
verdict = False
|
||||
break
|
||||
state[node] = 1
|
||||
path.append(node)
|
||||
if is_road[node]:
|
||||
verdict = True
|
||||
break
|
||||
following = int(receiver[node])
|
||||
if following == node or not valid[following]:
|
||||
verdict = False # 싱크에 갇히거나 해석 영역 밖으로 빠짐
|
||||
break
|
||||
node = following
|
||||
for visited in path:
|
||||
reaches[visited] = verdict
|
||||
state[visited] = 2
|
||||
resolved += len(path)
|
||||
return resolved
|
||||
|
||||
|
||||
def border_contact(active: np.ndarray) -> dict[str, bool]:
|
||||
"""활성 셀이 격자 최외곽에 닿은 방향. 전부 False면 유역이 능선 안에서 닫힌 것이다."""
|
||||
return {
|
||||
|
||||
@@ -353,16 +353,22 @@ def interpolate_elevation(spec: GridSpec, cloud: ContourCloud) -> np.ndarray:
|
||||
# ── ⑤ 웅덩이 채움 + 평탄면 해소 ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def condition_surface(surface: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
|
||||
def condition_surface(
|
||||
surface: np.ndarray, domain: np.ndarray | None = None
|
||||
) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""가짜 웅덩이를 채우고 평탄면에 미세 경사를 준다.
|
||||
|
||||
등고선 TIN은 같은 표고 정점 3개로 이루어진 평탄 삼각형과 계단형 가짜 웅덩이를
|
||||
필연적으로 만든다. 그대로 D8을 돌리면 흐름이 거기서 끊겨 상류 추적이 멈춘다.
|
||||
|
||||
채움은 형태학적 재구성(erosion)으로 한다. 배출구는 격자 최외곽과 TIN 경계(무효 셀에
|
||||
맞닿은 유효 셀)로 둔다 — 그래야 유효 영역 전체가 하나의 평탄면으로 잠기지 않는다.
|
||||
채움은 형태학적 재구성(erosion)으로 한다. 배출구는 격자 최외곽과 유효 영역 경계(무효
|
||||
셀에 맞닿은 유효 셀)로 둔다 — 그래야 유효 영역 전체가 하나의 평탄면으로 잠기지 않는다.
|
||||
|
||||
`domain`을 주면 그 안쪽만 해석 대상으로 삼는다(1차 영역에 걸쳐 실제 생성된 셀 마스크).
|
||||
"""
|
||||
valid = np.isfinite(surface)
|
||||
if domain is not None:
|
||||
valid &= domain
|
||||
if not valid.any():
|
||||
return surface, valid
|
||||
ceiling = float(np.nanmax(surface)) + 1000.0
|
||||
@@ -467,13 +473,18 @@ def compute_receivers(
|
||||
# ── 오케스트레이션 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_terrain_grid(spec: GridSpec, cloud: ContourCloud) -> TerrainGrid:
|
||||
"""격자 범위와 등고선 구름으로 지형 해석 격자를 만든다."""
|
||||
def build_terrain_grid(
|
||||
spec: GridSpec, cloud: ContourCloud, domain: np.ndarray | None = None
|
||||
) -> TerrainGrid:
|
||||
"""격자 범위와 등고선 구름으로 지형 해석 격자를 만든다.
|
||||
|
||||
`domain`은 실제 해석할 셀 마스크(1차 영역에 걸친 셀). 주면 그 밖은 무효로 둔다.
|
||||
"""
|
||||
surface = interpolate_elevation(spec, cloud)
|
||||
conditioned, valid = condition_surface(surface)
|
||||
conditioned, valid = condition_surface(surface, domain)
|
||||
receiver, step = compute_receivers(spec, conditioned, valid)
|
||||
logger.info(
|
||||
"배수유역: 격자 %d×%d (%.1fm), 유효 셀 %d개",
|
||||
"배수유역: 격자 %d×%d (%.2fm), 해석 대상 셀 %d개",
|
||||
spec.n_rows,
|
||||
spec.n_cols,
|
||||
spec.cell_m,
|
||||
@@ -484,6 +495,18 @@ def build_terrain_grid(spec: GridSpec, cloud: ContourCloud) -> TerrainGrid:
|
||||
)
|
||||
|
||||
|
||||
def direction_codes(spec: GridSpec, receiver: np.ndarray) -> np.ndarray:
|
||||
"""수신 셀 인덱스를 3×3 방향 코드로 바꾼다.
|
||||
|
||||
코드 = (행증분 + 1) * 3 + (열증분 + 1) → 0~8. 4는 제자리(싱크)를 뜻한다.
|
||||
화면이 셀마다 화살표를 그릴 수 있도록 방향만 뽑아낸 표현이다.
|
||||
"""
|
||||
index = np.arange(receiver.size, dtype=np.int64)
|
||||
row_delta = receiver // spec.n_cols - index // spec.n_cols
|
||||
col_delta = receiver % spec.n_cols - index % spec.n_cols
|
||||
return ((row_delta + 1) * 3 + (col_delta + 1)).astype(np.int8)
|
||||
|
||||
|
||||
def route_elevation_floor(route_z_values: list[float]) -> float | None:
|
||||
"""계획선 최저점에서 여유를 뺀 등고선 하한. 값이 없으면 None(필터 미적용)."""
|
||||
finite = [value for value in route_z_values if math.isfinite(value) and value != 0.0]
|
||||
|
||||
@@ -6,12 +6,14 @@
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import numpy as np
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
from pyproj import Transformer
|
||||
@@ -25,9 +27,9 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
|
||||
)
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Basin import (
|
||||
build_drainage_watershed,
|
||||
preview_primary_region,
|
||||
preview_stages,
|
||||
)
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Export import write_cell_mask, write_stage
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Export import write_grid_arrays, write_stage
|
||||
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import mask_row_spans
|
||||
from B05_wf2_Route.B05_wf2_Route_Repository import (
|
||||
get_latest_route,
|
||||
@@ -196,17 +198,18 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
prepared = await _prepare(project_id)
|
||||
if isinstance(prepared, JSONResponse):
|
||||
return prepared
|
||||
region = await asyncio.to_thread(
|
||||
preview_primary_region,
|
||||
preview = await asyncio.to_thread(
|
||||
preview_stages,
|
||||
prepared["vertices"],
|
||||
prepared["contours"],
|
||||
prepared["streams"],
|
||||
)
|
||||
if region is None:
|
||||
if preview is None:
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
content={"status": "error", "message": "1차 배수유역을 정할 등고선이 없습니다."},
|
||||
)
|
||||
region = preview.region
|
||||
to_lonlat = prepared["to_lonlat"]
|
||||
spec = region.spec
|
||||
payload = {
|
||||
@@ -240,6 +243,8 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
if region.cell_mask is not None
|
||||
else [],
|
||||
},
|
||||
# 셀별 흐름 방향과 도로 도달 여부. row_spans 순서(행 → 구간 → 열 오름차순)로 1바이트씩.
|
||||
"flow": _flow_payload(preview, region),
|
||||
}
|
||||
# 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
|
||||
payload["saved_to"] = write_stage(
|
||||
@@ -260,10 +265,72 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
||||
},
|
||||
to_lonlat,
|
||||
)
|
||||
write_cell_mask(prepared["stored_path"], "primary_region", spec, region.cell_mask)
|
||||
_write_stage_arrays(prepared["stored_path"], preview, region, spec)
|
||||
return payload
|
||||
|
||||
|
||||
def _write_stage_arrays(stored_path: str, preview: Any, region: Any, spec: Any) -> None:
|
||||
"""격자 규모 배열(셀 마스크·흐름 방향·도달 여부)을 단계별 `.npz`로 남긴다."""
|
||||
if region.cell_mask is not None:
|
||||
write_grid_arrays(
|
||||
stored_path,
|
||||
"primary_region",
|
||||
spec,
|
||||
{"mask": region.cell_mask},
|
||||
{"cells": region.active_cells, "bbox_cells": spec.size},
|
||||
)
|
||||
flow = preview.flow
|
||||
if flow is None:
|
||||
return
|
||||
write_grid_arrays(
|
||||
stored_path,
|
||||
"flow_direction",
|
||||
spec,
|
||||
{
|
||||
"direction": flow.direction.reshape(spec.n_rows, spec.n_cols),
|
||||
"reaches_road": flow.reaches_road.reshape(spec.n_rows, spec.n_cols),
|
||||
"analyzed": flow.analyzed.reshape(spec.n_rows, spec.n_cols),
|
||||
},
|
||||
{
|
||||
"analyzed": int(flow.analyzed.sum()),
|
||||
"reaches_road": int((flow.reaches_road & flow.analyzed).sum()),
|
||||
"no_road": int((~flow.reaches_road & flow.analyzed).sum()),
|
||||
"outer_seeds": flow.outer_seeds,
|
||||
"interior_seeds": flow.interior_seeds,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _flow_payload(preview: Any, region: Any) -> dict[str, Any] | None:
|
||||
"""셀별 흐름 방향·도로 도달 여부를 바이트 배열로 압축한다.
|
||||
|
||||
셀이 수십만 개라 JSON 객체로는 못 보낸다. 셀 하나당 1바이트로 줄이고 base64로 싣는다:
|
||||
하위 4비트 = 3×3 방향 코드(0~8, 4=제자리), 15 = 무효(표고 없음)
|
||||
최상위 비트(0x80) = 도로 도달(적색). 꺼져 있으면 미도달(파랑).
|
||||
바이트 순서는 `grid.row_spans`를 행 → 구간 → 열 오름차순으로 훑은 순서와 같다.
|
||||
"""
|
||||
flow = preview.flow
|
||||
if flow is None or region.cell_mask is None:
|
||||
return None
|
||||
order = np.flatnonzero(region.cell_mask.reshape(-1))
|
||||
codes = flow.direction[order]
|
||||
analyzed = flow.analyzed[order]
|
||||
reaches = flow.reaches_road[order]
|
||||
packed = np.where(codes < 0, 15, codes).astype(np.uint8)
|
||||
packed |= np.where(reaches, 0x80, 0).astype(np.uint8)
|
||||
return {
|
||||
"encoding": "base64-uint8",
|
||||
"cells": int(order.size),
|
||||
"reaches_road": int((reaches & analyzed).sum()),
|
||||
"no_road": int((~reaches & analyzed).sum()),
|
||||
# 격자에는 있으나 등고선 TIN 밖이라 표고가 없어 판정 못한 셀(방향 코드 15).
|
||||
"unanalyzed": int((~analyzed).sum()),
|
||||
"outer_seeds": flow.outer_seeds,
|
||||
"interior_seeds": flow.interior_seeds,
|
||||
"data": base64.b64encode(packed.tobytes()).decode("ascii"),
|
||||
}
|
||||
|
||||
|
||||
def _as_polygons(geometry: Any) -> list[Any]:
|
||||
if geometry is None or geometry.is_empty:
|
||||
return []
|
||||
|
||||
@@ -51,6 +51,18 @@ const LAYER_LABELS: Record<DrainageLayer, string> = {
|
||||
const ROUTE_COLOR = "#f97316";
|
||||
const COLLAPSED_KEY = "b05-route-drainage-collapsed";
|
||||
|
||||
// 해석 격자 셀 선 — 등고선·세류 위에 얹으므로 흰색으로 둔다(2026-07-31 사용자 지시).
|
||||
const GRID_LINE_COLOR = "rgba(255, 255, 255, 0.55)";
|
||||
// 흐름 판정 색 — 도로로 물이 오는 셀은 적색, 오지 않는 셀은 파랑.
|
||||
const FLOW_TO_ROAD_FILL = "rgba(220, 38, 38, 0.28)";
|
||||
const FLOW_TO_ROAD_LINE = "rgba(153, 27, 27, 0.95)";
|
||||
const FLOW_AWAY_FILL = "rgba(37, 99, 235, 0.22)";
|
||||
const FLOW_AWAY_LINE = "rgba(30, 64, 175, 0.9)";
|
||||
/** 등고선 TIN 밖이라 표고가 없어 판정하지 못한 셀 — 미도달(파랑)과 구분한다. */
|
||||
const FLOW_UNKNOWN_FILL = "rgba(120, 113, 108, 0.18)";
|
||||
/** 셀이 이보다 작으면 화살표가 뭉개져 읽히지 않으므로 채움색만 남긴다(px). */
|
||||
const ARROW_MIN_PX = 7;
|
||||
|
||||
/** 유역 오버레이 파스텔 색상. 번호 순으로 돌려쓴다(사용자 지시: 파스텔톤). */
|
||||
const BASIN_COLORS = [
|
||||
"rgba(167, 216, 199, 0.45)",
|
||||
@@ -159,6 +171,8 @@ export function createDrainagePanel(): DrainagePanel {
|
||||
// 1차 영역 검증 오버레이. null이면 표시하지 않는다.
|
||||
let primaryRegion: DrainagePrimaryRegion | null = null;
|
||||
let showRegion = false;
|
||||
// 흐름 방향 바이트 디코드 캐시 — 매 프레임 base64를 다시 풀지 않는다.
|
||||
let flowCache: { source: string; bytes: Uint8Array } | null = null;
|
||||
let scale = 1;
|
||||
let offsetX = 0;
|
||||
let offsetY = 0;
|
||||
@@ -272,11 +286,22 @@ export function createDrainagePanel(): DrainagePanel {
|
||||
context.stroke();
|
||||
}
|
||||
|
||||
/** 1차 영역에 걸쳐 실제로 생성된 셀만 사각형으로 그린다.
|
||||
/** 흐름 방향 바이트를 셀 순서대로 디코드한다(캐시 — 매 프레임 다시 풀지 않는다). */
|
||||
function flowBytes(region: DrainagePrimaryRegion): Uint8Array | null {
|
||||
if (!region.flow) return null;
|
||||
if (flowCache?.source === region.flow.data) return flowCache.bytes;
|
||||
const binary = atob(region.flow.data);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let i = 0; i < binary.length; i += 1) bytes[i] = binary.charCodeAt(i);
|
||||
flowCache = { source: region.flow.data, bytes };
|
||||
return bytes;
|
||||
}
|
||||
|
||||
/** 1차 영역에 걸쳐 실제로 생성된 셀만 그린다.
|
||||
*
|
||||
* bbox 전체를 채우지 않는다 — 백엔드가 준 행별 구간(row_spans)만 그린다.
|
||||
* 셀이 화면에서 2px 미만이면 선이 뭉개져 회색 덩어리가 되므로, 그때는 구간을 통짜
|
||||
* 사각형으로 채워 격자가 덮은 범위만 흐리게 보여 준다(확대하면 셀 하나하나가 보인다). */
|
||||
* bbox 전체를 채우지 않는다 — 백엔드가 준 행별 구간(row_spans)만 그린다. 흐름 판정이
|
||||
* 있으면 셀마다 방향 화살표를 얹고, 도로에 물이 닿는 셀은 적색·닿지 않으면 파랑으로
|
||||
* 칠한다. 셀이 화면에서 작아지면 화살표가 안 보이므로 채움색만 남긴다. */
|
||||
function drawGridCells(
|
||||
context: CanvasRenderingContext2D,
|
||||
map: Normalizer,
|
||||
@@ -299,33 +324,115 @@ export function createDrainagePanel(): DrainagePanel {
|
||||
const { rows, cols, row_spans: spans } = region.grid;
|
||||
const cellW = (right - left) / Math.max(cols, 1);
|
||||
const cellH = (bottom - top) / Math.max(rows, 1);
|
||||
const fine = Math.min(Math.abs(cellW), Math.abs(cellH)) >= 2;
|
||||
const cellPx = Math.min(Math.abs(cellW), Math.abs(cellH));
|
||||
const bytes = flowBytes(region);
|
||||
|
||||
context.save();
|
||||
context.setLineDash([]);
|
||||
context.strokeStyle = "rgba(120, 113, 108, 0.3)";
|
||||
context.fillStyle = "rgba(120, 113, 108, 0.12)";
|
||||
context.lineWidth = 0.5;
|
||||
if (fine) context.beginPath();
|
||||
context.lineCap = "round";
|
||||
let cursor = 0; // row_spans를 훑은 순서 = 흐름 바이트 순서
|
||||
spans.forEach(([row, colStart, colEnd]) => {
|
||||
const count = colEnd - colStart + 1;
|
||||
const base = cursor;
|
||||
cursor += count;
|
||||
const y = top + cellH * row;
|
||||
if (y + cellH < -50 || y > view.height + 50) return;
|
||||
if (y + cellH < -40 || y > view.height + 40) return;
|
||||
const x = left + cellW * colStart;
|
||||
const width = cellW * (colEnd - colStart + 1);
|
||||
if (x + width < -50 || x > view.width + 50) return;
|
||||
if (!fine) {
|
||||
// 축소 상태 — 구간을 통짜로 칠해 격자가 덮은 범위만 흐리게 보여 준다.
|
||||
context.fillRect(x, y, width, cellH);
|
||||
const width = cellW * count;
|
||||
if (x + width < -40 || x > view.width + 40) return;
|
||||
|
||||
if (!bytes) {
|
||||
// 흐름 판정 전 — 격자만 흰 선으로 보여 준다.
|
||||
if (cellPx >= 2) {
|
||||
context.strokeStyle = GRID_LINE_COLOR;
|
||||
context.lineWidth = 0.5;
|
||||
context.beginPath();
|
||||
for (let col = colStart; col <= colEnd; col += 1) {
|
||||
context.rect(left + cellW * col, y, cellW, cellH);
|
||||
}
|
||||
context.stroke();
|
||||
} else {
|
||||
context.fillStyle = "rgba(255, 255, 255, 0.2)";
|
||||
context.fillRect(x, y, width, cellH);
|
||||
}
|
||||
return;
|
||||
}
|
||||
for (let col = colStart; col <= colEnd; col += 1) {
|
||||
context.rect(left + cellW * col, y, cellW, cellH);
|
||||
for (let offset = 0; offset < count; offset += 1) {
|
||||
drawFlowCell(
|
||||
context,
|
||||
bytes[base + offset],
|
||||
left + cellW * (colStart + offset),
|
||||
y,
|
||||
cellW,
|
||||
cellH,
|
||||
cellPx,
|
||||
);
|
||||
}
|
||||
});
|
||||
if (fine) context.stroke();
|
||||
context.restore();
|
||||
}
|
||||
|
||||
/** 셀 하나 — 도달 여부로 칠하고, 충분히 크면 흐름 방향 화살표를 얹는다. */
|
||||
function drawFlowCell(
|
||||
context: CanvasRenderingContext2D,
|
||||
code: number,
|
||||
x: number,
|
||||
y: number,
|
||||
cellW: number,
|
||||
cellH: number,
|
||||
cellPx: number,
|
||||
): void {
|
||||
const direction = code & 0x0f;
|
||||
const reaches = (code & 0x80) !== 0;
|
||||
// 코드 15 = 등고선 TIN 밖이라 표고가 없어 판정 못한 셀. 파랑(미도달)과 구분한다.
|
||||
const unanalyzed = direction === 15;
|
||||
context.fillStyle = unanalyzed
|
||||
? FLOW_UNKNOWN_FILL
|
||||
: reaches
|
||||
? FLOW_TO_ROAD_FILL
|
||||
: FLOW_AWAY_FILL;
|
||||
context.fillRect(x, y, cellW, cellH);
|
||||
if (cellPx >= 2) {
|
||||
context.strokeStyle = GRID_LINE_COLOR;
|
||||
context.lineWidth = 0.5;
|
||||
context.strokeRect(x, y, cellW, cellH);
|
||||
}
|
||||
if (cellPx < ARROW_MIN_PX || direction === 15) return;
|
||||
const stroke = reaches ? FLOW_TO_ROAD_LINE : FLOW_AWAY_LINE;
|
||||
const midX = x + cellW / 2;
|
||||
const midY = y + cellH / 2;
|
||||
if (direction === 4) {
|
||||
// 제자리(싱크) — 방향이 없으므로 점으로 표시한다.
|
||||
context.fillStyle = stroke;
|
||||
context.beginPath();
|
||||
context.arc(midX, midY, Math.max(1, cellPx * 0.12), 0, Math.PI * 2);
|
||||
context.fill();
|
||||
return;
|
||||
}
|
||||
const colDelta = (direction % 3) - 1;
|
||||
const rowDelta = Math.floor(direction / 3) - 1;
|
||||
const length = Math.hypot(colDelta, rowDelta) || 1;
|
||||
const reach = (cellPx * 0.38) / length;
|
||||
const tipX = midX + colDelta * reach;
|
||||
const tipY = midY + rowDelta * reach;
|
||||
context.strokeStyle = stroke;
|
||||
context.lineWidth = Math.max(0.6, cellPx * 0.09);
|
||||
context.beginPath();
|
||||
context.moveTo(midX - colDelta * reach, midY - rowDelta * reach);
|
||||
context.lineTo(tipX, tipY);
|
||||
context.stroke();
|
||||
// 촉 — 진행 방향 기준 좌우로 짧게 접는다.
|
||||
const head = cellPx * 0.16;
|
||||
const unitX = (colDelta / length) * head;
|
||||
const unitY = (rowDelta / length) * head;
|
||||
context.beginPath();
|
||||
context.moveTo(tipX, tipY);
|
||||
context.lineTo(tipX - unitX - unitY * 0.7, tipY - unitY + unitX * 0.7);
|
||||
context.moveTo(tipX, tipY);
|
||||
context.lineTo(tipX - unitX + unitY * 0.7, tipY - unitY - unitX * 0.7);
|
||||
context.stroke();
|
||||
}
|
||||
|
||||
/** 1차 배수유역 근거를 겹쳐 그린다 — 단계 검증용. */
|
||||
function drawPrimaryRegion(
|
||||
context: CanvasRenderingContext2D,
|
||||
@@ -513,12 +620,20 @@ export function createDrainagePanel(): DrainagePanel {
|
||||
const cells = region.grid.cells.toLocaleString();
|
||||
const outside =
|
||||
region.road_outside_m > 0 ? ` · 노선 이탈 ${Math.round(region.road_outside_m)}m` : "";
|
||||
const unknown =
|
||||
region.flow && region.flow.unanalyzed > 0
|
||||
? ` / 표고없음 ${region.flow.unanalyzed.toLocaleString()}(회)`
|
||||
: "";
|
||||
const flow = region.flow
|
||||
? ` · 흐름 도로도달 ${region.flow.reaches_road.toLocaleString()}(적) / ` +
|
||||
`미도달 ${region.flow.no_road.toLocaleString()}(청)${unknown}, ` +
|
||||
`최외곽 출발 ${region.flow.outer_seeds.toLocaleString()} + ` +
|
||||
`내부 보충 ${region.flow.interior_seeds.toLocaleString()}`
|
||||
: " · 흐름 판정 없음";
|
||||
return (
|
||||
`1차 영역(반경 ${region.radius_m}m): 상류망 ${region.upstream_lines.length}조각 채택 / ` +
|
||||
`하류망 ${region.downstream_lines.length}조각·미연결 ${region.no_contact_count}개 제외 · ` +
|
||||
`격자 ${region.grid.cell_m}m(도로 시점 기준) 셀 ${cells}개 ` +
|
||||
`/ bbox ${region.grid.width_m}×${region.grid.height_m}m ` +
|
||||
`${region.grid.bbox_cells.toLocaleString()}셀${outside}`
|
||||
`격자 ${region.grid.cell_m}m(도로 시점 기준) 셀 ${cells}개${outside}${flow}`
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user