feat(B05): 흐름 방향을 등고선 하강 방식으로 교체 (TIN 보간 폐기)

영구저장소 데이터를 오프라인 재현해 원인을 특정한 뒤 방식을 바꿨다.

진단 (실데이터 536,708셀, 저장본과 정확히 일치 재현)
- 파랑 셀 365,303개의 사슬 종료 사유가 100% 싱크. 영역 이탈 0%.
- 싱크 3,974개의 표고가 전부 5의 배수(600.0/765.0/585.0) = 등고선 값.
- 원인: 등고선 TIN 보간이 만든 평탄면을 EDT 로 해소할 때 가장 가까운
  비평탄 셀을 출구로 삼는데, 그게 오르막이면 물이 나갈 수 없어 싱크로 남는다.
  싱크 하나가 상류 유역 전체를 삼켰다(최대 10,171셀).
- 화살표(기울기 32방위)와 추적(D8)이 서로 달라 2,055셀에서 어긋났다.

교체 방식 — Watershed_Descent.py 신설
  1. 등고선을 격자에 직접 굽는다(보간면을 만들지 않는다)
  2. 셀마다 가장 가까운 등고 라인의 표고를 밴드로 삼는다
  3. 높은 밴드부터 내려오며 한 단 낮은 등고 라인까지 거리를 잰다
  4. 위치에너지 = 밴드 순위 x 큰 수 + 그 거리
  5. 수신 셀 = 위치에너지가 더 낮은 8이웃 중 화살표 방향에 가장 가까운 셀

위치에너지가 흐름을 따라 반드시 감소하므로 웅덩이도 순환도 원리적으로
생기지 않는다 — 채움/평탄해소 자체가 불필요해졌다. 수신 셀을 화살표에서
고르므로 화면 화살표와 실제 경로가 항상 일치한다.

실측 (같은 데이터)
              적색            파랑           싱크    화살표=경로
  TIN 방식    171,405 (32%)  365,303 (68%)  3,974   불일치 2,055셀
  등고선 하강 424,193 (79%)  112,515 (21%)  15,911  일치 100%, 불일치 0셀

부수 수정
- 최하단 밴드 셀을 무효가 아닌 정지 셀로 남겨 도로/세류면 적색으로 잡히게 함
- 저장에 receiver / band_elevation 추가 (사후에 사슬을 다시 따라갈 수 있게)
- manifest 에 row_spans 가 통째로 들어가 110KB 가 되던 것 정리

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-31 19:23:33 +09:00
co-authored by Claude Fable 5
parent 12bee4bc82
commit 445f51bbcb
4 changed files with 334 additions and 27 deletions
@@ -35,6 +35,10 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
find_stream_crossings,
is_uphill_at,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Descent import (
ContourDescent,
build_contour_descent,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Flow import (
FlowClassification,
RoadRaster,
@@ -50,7 +54,6 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
GridSpec,
TerrainGrid,
build_contour_cloud,
build_terrain_grid,
route_elevation_floor,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Stream import (
@@ -58,7 +61,6 @@ 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,
@@ -215,6 +217,7 @@ class StagePreview:
terrain: TerrainGrid | None = None
road: RoadRaster | None = None
flow: FlowClassification | None = None
descent: ContourDescent | None = None
def preview_stages(
@@ -224,8 +227,12 @@ def preview_stages(
) -> StagePreview | None:
"""지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다.
현재 포함: ① 1차 배수유역 ② 격자 생성 ③ 표고·D8 ④ 흐름 방향/도로 도달 판정.
**격자 확장은 넣지 않는다** — 다음 검증 단계다(2026-07-31 사용자 지시).
현재 포함: ① 1차 배수유역 ② 격자 생성 ③ **등고선 하강 방향** ④ 도로 도달 판정.
③은 보간면(TIN)을 쓰지 않는다. 높은 등고 라인에서 낮은 등고 라인으로 내려가며 방향을
세우므로 가짜 웅덩이·평탄면이 원리적으로 생기지 않는다(2026-07-31 사용자 지시로 방식 교체).
**격자 확장은 넣지 않는다** — 다음 검증 단계다.
"""
if len(vertices) < 2:
return None
@@ -234,31 +241,29 @@ def preview_stages(
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("배수유역: 격자 범위 안에 등고선이 없어 흐름 판정을 건너뜁니다.")
started = time.perf_counter()
floor = route_elevation_floor([vertex.z for vertex in vertices])
descent = build_contour_descent(spec, contour_features, region.cell_mask, floor)
if not descent.valid.any():
logger.warning("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.")
return StagePreview(region=region)
started = time.perf_counter()
terrain = build_terrain_grid(spec, cloud, region.cell_mask)
# 이후 단계(유역 제원)가 표고를 쓰므로 밴드 표고를 지형 격자로 함께 들고 간다.
terrain = TerrainGrid(
spec=spec,
elevation=descent.band_elevation,
valid=descent.valid,
receiver=descent.receiver,
step_length=descent.step_length,
)
road = rasterize_road(spec, route_line, DRAINAGE_ROAD_WIDTH_M)
# 확정된 상류 세류망을 따라 흐름을 새긴다 — 세류선 위 셀과 그리로 흘러드는 셀은
# 반드시 도로에 도달해야 한다(TIN 보간면은 실제 물골을 재현하지 못한다).
# 반드시 도로에 도달해야 한다.
terrain, burned = burn_stream_flow(terrain, road, region.split.upstream)
flow = classify_flow(terrain, road, burned)
flow = classify_flow(terrain, road, burned, azimuth=descent.azimuth)
logger.info("배수유역: 흐름 판정 %.1fs (셀 %d개)", time.perf_counter() - started, spec.size)
return StagePreview(region=region, terrain=terrain, road=road, flow=flow)
return StagePreview(region=region, terrain=terrain, road=road, flow=flow, descent=descent)
# ── ③~④ 격자 해석 (캐시 대상) ───────────────────────────────────────────────
@@ -0,0 +1,265 @@
"""등고선 기반 흐름 방향 — 높은 등고 라인에서 낮은 등고 라인으로 내려가며 방향을 세운다.
기존 방식(등고선 → TIN 보간 → 지표면 기울기 → D8)은 보간면이 만든 가짜 웅덩이와 평탄
삼각형 때문에 흐름이 중간에서 끊겼다. 실데이터에서 채움·평탄해소를 거치고도 싱크가 수천 개
남았고, 그 싱크 하나가 상류 유역 전체를 통째로 삼켰다.
여기서는 **보간면을 거치지 않는다.** 등고선을 격자에 직접 굽고, 셀마다 "내가 속한 등고
라인보다 한 단 낮은 등고 라인"이 어디인지를 찾아 그쪽으로 방향을 준다(2026-07-31 사용자 지시).
① 등고선을 격자에 굽는다 — 셀이 어느 표고의 라인 위인지 기록
② 셀마다 가장 가까운 등고 라인을 찾아 그 표고를 '밴드'로 삼는다
③ 표고가 높은 밴드부터 내려오며, 각 밴드에서 **한 단 낮은 등고 라인까지의 거리**를 잰다
④ 위치에너지 = 밴드 순위 × 큰 수 + 그 거리
⑤ 수신 셀 = 위치에너지가 더 낮은 8이웃 중 화살표 방향에 가장 가까운 셀
④의 위치에너지는 흐름을 따라 반드시 감소한다. 그래서 **순환도 웅덩이도 원리적으로 생기지
않는다** — 채움이나 평탄면 해소가 아예 필요 없다.
⑤ 덕분에 화면 화살표(32방위)와 실제 추적 경로가 항상 같은 방향을 가리킨다. 예전에는
화살표는 기울기, 추적은 D8이라 서로 어긋나 눈으로 검증할 수가 없었다.
"""
from __future__ import annotations
import logging
import math
from dataclasses import dataclass
from typing import Any
import numpy as np
from rasterio.features import rasterize
from scipy.ndimage import distance_transform_edt
from shapely.geometry import shape
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
AZIMUTH_INVALID,
AZIMUTH_SINK,
AZIMUTH_STEPS,
GridSpec,
_feature_elevation,
grid_transform,
iter_linestrings,
)
from config.config_system import DRAINAGE_CONTOUR_MIN_LENGTH_M
logger = logging.getLogger(__name__)
# 8이웃 (행 증분, 열 증분).
_NEIGHBOURS = (
(-1, -1),
(-1, 0),
(-1, 1),
(0, -1),
(0, 1),
(1, -1),
(1, 0),
(1, 1),
)
@dataclass
class ContourDescent:
"""등고선에서 직접 세운 흐름 방향 격자."""
spec: GridSpec
band_elevation: np.ndarray # (R, C) float32 — 셀이 속한 등고 라인 표고, 무효는 NaN
valid: np.ndarray # (R, C) bool — 방향을 세운 셀
receiver: np.ndarray # (R*C,) int32 — 다음 셀, 최하단 밴드는 자기 자신
step_length: np.ndarray # (R*C,) float32
azimuth: np.ndarray # (R*C,) int16 — 32방위 코드(32=제자리, 33=무효)
levels: list[float] # 사용된 등고 표고(내림차순)
def rasterize_contours(
spec: GridSpec,
contour_features: list[dict[str, Any]],
elevation_floor_m: float | None = None,
) -> tuple[np.ndarray, list[float]]:
"""등고선을 격자에 굽는다. 셀마다 그 위를 지나는 등고 라인의 표고(없으면 NaN)."""
by_level: dict[float, list[Any]] = {}
for feature in contour_features:
geometry = feature.get("geometry")
if not geometry:
continue
elevation = _feature_elevation(feature.get("properties") or {})
if elevation is None:
continue
if elevation_floor_m is not None and elevation < elevation_floor_m:
continue
try:
parsed = shape(geometry)
except Exception: # noqa: BLE001
continue
for line in iter_linestrings(parsed):
if line.length < DRAINAGE_CONTOUR_MIN_LENGTH_M:
continue
by_level.setdefault(float(elevation), []).append(line)
burned = np.full((spec.n_rows, spec.n_cols), np.nan, dtype=np.float32)
levels = sorted(by_level, reverse=True)
transform = grid_transform(spec)
for elevation in levels:
stamp = rasterize(
[(line, 1) for line in by_level[elevation]],
out_shape=(spec.n_rows, spec.n_cols),
transform=transform,
fill=0,
dtype="uint8",
all_touched=True,
).astype(bool)
# 낮은 표고부터 덮어써야 겹치는 셀이 낮은 라인으로 남는다 — 물은 낮은 쪽으로 간다.
burned[stamp] = elevation
logger.info(
"배수유역: 등고 라인 %d단(%.0f~%.0fm)을 격자에 굽어 %d",
len(levels),
levels[-1] if levels else 0.0,
levels[0] if levels else 0.0,
int(np.isfinite(burned).sum()),
)
return burned, levels
def build_contour_descent(
spec: GridSpec,
contour_features: list[dict[str, Any]],
domain: np.ndarray | None = None,
elevation_floor_m: float | None = None,
) -> ContourDescent:
"""등고선만으로 셀별 흐름 방향을 세운다. 보간면을 만들지 않는다."""
rows, cols = spec.n_rows, spec.n_cols
burned, levels = rasterize_contours(spec, contour_features, elevation_floor_m)
empty = ContourDescent(
spec=spec,
band_elevation=np.full((rows, cols), np.nan, dtype=np.float32),
valid=np.zeros((rows, cols), dtype=bool),
receiver=np.arange(spec.size, dtype=np.int32),
step_length=np.zeros(spec.size, dtype=np.float32),
azimuth=np.full(spec.size, AZIMUTH_INVALID, dtype=np.int16),
levels=levels,
)
if len(levels) < 2:
logger.warning("배수유역: 등고 라인이 2단 미만이라 방향을 세울 수 없습니다.")
return empty
on_contour = np.isfinite(burned)
# ② 셀마다 가장 가까운 등고 라인의 표고 = 그 셀의 밴드.
_, (near_row, near_col) = distance_transform_edt(~on_contour, return_indices=True)
band_elevation = burned[near_row, near_col].astype(np.float32)
inside = domain if domain is not None else np.ones((rows, cols), dtype=bool)
band_elevation = np.where(inside, band_elevation, np.nan)
# ③④ 높은 밴드부터 내려오며 한 단 낮은 등고 라인까지의 거리와 목표 셀을 구한다.
distance = np.full((rows, cols), np.inf, dtype=np.float32)
target_row = np.zeros((rows, cols), dtype=np.int32)
target_col = np.zeros((rows, cols), dtype=np.int32)
band_rank = np.full((rows, cols), -1, dtype=np.int32)
for rank, elevation in enumerate(levels[:-1]):
members = inside & (band_elevation == elevation)
if not members.any():
continue
lower = on_contour & (burned < elevation)
if not lower.any():
continue
step_distance, (step_row, step_col) = distance_transform_edt(~lower, return_indices=True)
distance[members] = step_distance[members].astype(np.float32)
target_row[members] = step_row[members]
target_col[members] = step_col[members]
band_rank[members] = len(levels) - 1 - rank # 높을수록 큰 값
# 최하단 밴드는 더 내려갈 등고 라인이 없다. 무효로 빼지 않고 **정지 셀**로 남긴다 —
# 그래야 그 자리가 도로·세류선이면 적색으로 잡히고, 아니면 물이 고이는 지점으로 보인다.
lowest = inside & (band_elevation == levels[-1]) & (band_rank < 0)
if lowest.any():
distance[lowest] = 0.0
band_rank[lowest] = 0
valid = band_rank >= 0
if not valid.any():
logger.warning("배수유역: 하강 방향을 세운 셀이 없습니다.")
return empty
# 밴드가 하나 낮아지면 위치에너지가 반드시 떨어지도록 거리 최대치보다 큰 간격을 준다.
finite = distance[valid & np.isfinite(distance)]
span = (float(finite.max()) if finite.size else 1.0) + 2.0
distance[valid & ~np.isfinite(distance)] = 0.0
potential = np.where(valid, band_rank.astype(np.float64) * span + distance, np.inf)
receiver, step_length, azimuth = _route_by_potential(
spec, potential, valid, target_row, target_col
)
logger.info(
"배수유역: 등고선 하강 방향 %d셀 (밴드 %d단), 최하단 정지 %d",
int(valid.sum()),
int(band_rank[valid].max() - band_rank[valid].min() + 1),
int((azimuth == AZIMUTH_SINK).sum()),
)
return ContourDescent(
spec=spec,
band_elevation=np.where(valid, band_elevation, np.nan).astype(np.float32),
valid=valid,
receiver=receiver,
step_length=step_length,
azimuth=azimuth,
levels=levels,
)
def _route_by_potential(
spec: GridSpec,
potential: np.ndarray,
valid: np.ndarray,
target_row: np.ndarray,
target_col: np.ndarray,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""위치에너지가 낮은 8이웃 중 **화살표 방향에 가장 가까운** 셀을 수신 셀로 고른다.
화살표는 "한 단 낮은 등고 라인 쪽"을 가리키는 연속 방위이고, 수신 셀은 그 방위에 가장
가까운 이웃이다. 그래서 화면 화살표와 실제 추적 경로가 어긋나지 않는다.
위치에너지가 더 낮은 이웃만 후보로 두므로 순환이 생기지 않는다.
"""
rows, cols = spec.n_rows, spec.n_cols
grid_row, grid_col = np.meshgrid(np.arange(rows), np.arange(cols), indexing="ij")
# 목표(한 단 낮은 등고 라인 위의 셀)를 향하는 연속 방위.
aim_row = (target_row - grid_row).astype(np.float64)
aim_col = (target_col - grid_col).astype(np.float64)
aim_norm = np.hypot(aim_row, aim_col)
aim_norm[aim_norm == 0.0] = 1.0
aim_row /= aim_norm
aim_col /= aim_norm
padded = np.full((rows + 2, cols + 2), np.inf)
padded[1:-1, 1:-1] = potential
flat_index = np.arange(spec.size, dtype=np.int32).reshape(rows, cols)
padded_index = np.full((rows + 2, cols + 2), -1, dtype=np.int32)
padded_index[1:-1, 1:-1] = flat_index
best_score = np.full((rows, cols), -np.inf)
receiver = flat_index.copy()
step = np.zeros((rows, cols), dtype=np.float32)
centre = potential
for row_shift, col_shift in _NEIGHBOURS:
neighbour = padded[
1 + row_shift : 1 + row_shift + rows, 1 + col_shift : 1 + col_shift + cols
]
length = math.hypot(row_shift, col_shift)
# 방위 일치도(코사인 유사도)가 클수록 좋은 후보다.
score = (aim_row * row_shift + aim_col * col_shift) / length
better = valid & np.isfinite(neighbour) & (neighbour < centre) & (score > best_score)
if not better.any():
continue
best_score = np.where(better, score, best_score)
neighbour_index = padded_index[
1 + row_shift : 1 + row_shift + rows, 1 + col_shift : 1 + col_shift + cols
]
receiver = np.where(better, neighbour_index, receiver)
step = np.where(better, np.float32(length * spec.cell_m), step)
moved = receiver != flat_index
delta_row = (receiver // cols - flat_index // cols).astype(np.float64)
delta_col = (receiver % cols - flat_index % cols).astype(np.float64)
angle = np.arctan2(delta_row, delta_col)
code = np.rint(angle / (2.0 * math.pi / AZIMUTH_STEPS)).astype(np.int16) % AZIMUTH_STEPS
azimuth = np.where(moved, code, AZIMUTH_SINK)
azimuth = np.where(valid, azimuth, AZIMUTH_INVALID)
return receiver.reshape(-1), step.reshape(-1), azimuth.reshape(-1).astype(np.int16)
@@ -25,6 +25,7 @@ from shapely.geometry import LineString, MultiPolygon, Polygon, shape
from shapely.ops import unary_union
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
AZIMUTH_STEPS,
ContourCloud,
GridSpec,
TerrainGrid,
@@ -297,7 +298,10 @@ def outermost_cells(domain: np.ndarray) -> np.ndarray:
def classify_flow(
terrain: TerrainGrid, road: RoadRaster, burned: np.ndarray | None = None
terrain: TerrainGrid,
road: RoadRaster,
burned: np.ndarray | None = None,
azimuth: np.ndarray | None = None,
) -> FlowClassification:
"""최외곽 셀부터 물길을 따라가며 도로 도달 여부를 판정한다.
@@ -317,8 +321,9 @@ def classify_flow(
다시 분석하지 않는다.
④ 최외곽 추적에 안 걸린 내부 셀을 그다음에 따로 출발시킨다.
화살표 방위는 D8(8방위)이 아니라 지형 최급강하 32방위를 쓴다. 세류 셀은 확정된 물길
방향을 그대로 쓴다.
`azimuth`를 주면 그 32방위 코드를 그대로 화살표로 쓴다(등고선 하강 방향). 주지 않으면
지표면 기울기에서 뽑는다. **화살표는 실제 수신 셀과 같은 방향이어야 한다** — 어긋나면
화살표로 사슬을 따라가는 눈 검증이 성립하지 않는다.
"""
spec = terrain.spec
valid = terrain.valid.reshape(-1)
@@ -327,7 +332,10 @@ def classify_flow(
stream_cells = np.zeros(spec.size, dtype=bool) if burned is None else burned
absorbing = (road.mask.reshape(-1) & valid) | stream_cells
direction = descent_azimuth(spec, terrain.elevation, terrain.valid, receiver, burned)
if azimuth is None:
direction = descent_azimuth(spec, terrain.elevation, terrain.valid, receiver, burned)
else:
direction = _azimuth_with_burn(spec, azimuth, receiver, burned)
reaches = np.zeros(spec.size, dtype=bool)
# 0=미방문, 1=경로에 올라 있음, 2=판정 완료
state = np.zeros(spec.size, dtype=np.int8)
@@ -363,6 +371,26 @@ def classify_flow(
)
def _azimuth_with_burn(
spec: GridSpec, azimuth: np.ndarray, receiver: np.ndarray, burned: np.ndarray | None
) -> np.ndarray:
"""세류망을 따라 흐름을 새긴 셀은 그 수신 셀 방향으로 화살표를 덮어쓴다."""
direction = azimuth.astype(np.int16, copy=True)
if burned is None or not burned.any():
return direction
index = np.arange(spec.size, dtype=np.int64)
moved = burned & (receiver != index)
if not moved.any():
return direction
row_delta = (receiver[moved] // spec.n_cols - index[moved] // spec.n_cols).astype(np.float64)
col_delta = (receiver[moved] % spec.n_cols - index[moved] % spec.n_cols).astype(np.float64)
angle = np.arctan2(row_delta, col_delta)
direction[moved] = (
np.rint(angle / (2.0 * np.pi / AZIMUTH_STEPS)).astype(np.int16) % AZIMUTH_STEPS
)
return direction
def _walk_from(
starts: np.ndarray,
receiver: np.ndarray,
+10 -1
View File
@@ -266,7 +266,12 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
"radius_m": region.radius_m,
"road_outside_m": payload["road_outside_m"],
"no_contact_count": region.split.no_contact,
"grid": {key: value for key, value in payload["grid"].items() if key != "bbox_lonlat"},
# 기하(bbox·셀 구간)는 파일 본문에 있으므로 요약 수치만 남긴다.
"grid": {
key: value
for key, value in payload["grid"].items()
if key not in {"bbox_lonlat", "row_spans"}
},
},
to_lonlat,
)
@@ -291,9 +296,13 @@ def _write_stage_arrays(stored_path: str, preview: Any, region: Any, spec: Any)
"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),
# 사후 진단용 — 수신 셀이 있어야 사슬을 다시 따라가 볼 수 있다.
"receiver": preview.terrain.receiver.reshape(spec.n_rows, spec.n_cols),
}
if flow.burned is not None:
arrays["burned"] = flow.burned.reshape(spec.n_rows, spec.n_cols)
if preview.descent is not None:
arrays["band_elevation"] = preview.descent.band_elevation
write_grid_arrays(
stored_path,
"flow_direction",