feat(B05): 최외곽 적색 셀 주변 확장 루프 (Watershed_Expand 신설)

적색 셀이 해석 영역 최외곽에 있다는 것은 그 바깥에서 물이 더 들어온다는 뜻이다.
거기서 멈추면 유역이 잘린다. 반대로 최외곽이 전부 파랑이면 바깥 물은 도로로
오지 않으므로 더 볼 필요가 없다.

  1. 해석 영역 최외곽 셀 중 적색인 것을 찾는다
  2. 그 주변으로 한 겹(기본 50m) 넓힌다
  3. 넓힌 영역으로 흐름 방향/색을 다시 분석한다
  4. 새로 추가한 셀에 적색이 없으면 종료 (사용자 지시)
     최외곽에 적색이 아예 없어도 종료 - 둘 다 같은 판단이다

격자 bbox 에 닿으면 격자도 셀 정수배로 넓힌다. 도로 시작점 기준 격자점은
그대로 유지되므로 확장 전후 같은 자리 셀이 같은 자리에 남는다.

실데이터 결과 (33.3s)
  최외곽 적색  1,184 -> 400 -> 16 -> 0 (3회차에 닫힘)
  해석 셀      536,708 -> 640,872 (+104,164)
  적색         424,193 -> 457,404
  화살표=경로 일치 100%, 파랑인데 적색 가리킴 0개 (유지)

구조 변경
- Watershed_Expand.py 신설: analyze_domain(1회 분석) + expand_by_red_boundary(루프)
- StagePreview 에 spec/domain/expand_* 추가. 확장하면 격자가 1차 영역보다
  커지므로 화면/저장은 region.spec 이 아니라 이쪽을 봐야 한다.
- 라우터 응답에 expansion(회차/닫힘/추가 셀 수) 추가
- config: DRAINAGE_RED_EXPAND_BAND_M=50, DRAINAGE_RED_EXPAND_MAX_ROUNDS=20

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-31 19:36:38 +09:00
co-authored by Claude Fable 5
parent 445f51bbcb
commit 44eb4ad2af
6 changed files with 311 additions and 42 deletions
+9
View File
@@ -327,6 +327,15 @@ export interface DrainagePrimaryRegion {
/** 실제 생성된 셀 구간 [행, 시작열, 끝열(포함)]. 낱개 셀 대신 구간으로 온다. */
row_spans: Array<[number, number, number]>;
};
/** 최외곽 적색 셀 주변 확장 결과. */
expansion: {
rounds: number;
/** 새로 추가한 셀에 적색이 없어 스스로 멈췄는가. */
closed: boolean;
added_cells: number;
/** 확장 전(1차 영역) 셀 수. */
initial_cells: number;
};
/** 셀별 흐름 방향과 도로 도달 여부. 등고선이 없어 판정을 못하면 null. */
flow: {
encoding: "base64-uint8";
@@ -35,20 +35,15 @@ 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_Descent import ContourDescent
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Expand import expand_by_red_boundary
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Flow import (
FlowClassification,
RoadRaster,
burn_stream_flow,
classify_flow,
expand_until_closed,
largest_ring,
outer_boundary,
polygonize_labels,
rasterize_road,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
GridSpec,
@@ -211,13 +206,22 @@ def preview_primary_region(
@dataclass
class StagePreview:
"""단계 검증 산출물 묶음. 기능을 붙일 때마다 여기에 항목이 하나씩 늘어난다."""
"""단계 검증 산출물 묶음. 기능을 붙일 때마다 여기에 항목이 하나씩 늘어난다.
확장을 거치면 격자와 해석 영역이 1차 영역보다 커진다. 화면·저장은 `region.spec`이
아니라 여기 `spec`/`domain`을 봐야 한다.
"""
region: PrimaryRegion
spec: GridSpec | None = None
domain: np.ndarray | None = None
terrain: TerrainGrid | None = None
road: RoadRaster | None = None
flow: FlowClassification | None = None
descent: ContourDescent | None = None
expand_rounds: int = 0
expand_closed: bool = False
expand_added_cells: int = 0
def preview_stages(
@@ -227,12 +231,14 @@ def preview_stages(
) -> StagePreview | None:
"""지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다.
현재 포함: ① 1차 배수유역 ② 격자 생성 ③ **등고선 하강 방향** ④ 도로 도달 판정.
현재 포함: ① 1차 배수유역 ② 격자 생성 ③ **등고선 하강 방향** ④ 도로 도달 판정
⑤ **최외곽 적색 셀 주변 확장**.
③은 보간면(TIN)을 쓰지 않는다. 높은 등고 라인에서 낮은 등고 라인으로 내려가며 방향을
세우므로 가짜 웅덩이·평탄면이 원리적으로 생기지 않는다(2026-07-31 사용자 지시로 방식 교체).
**격자 확장은 넣지 않는다** — 다음 검증 단계다.
⑤는 최외곽에 적색이 남아 있으면 그 주변으로 넓혀 다시 분석하고, **새로 추가한 셀에
적색이 없으면** 멈춘다.
"""
if len(vertices) < 2:
return None
@@ -241,29 +247,39 @@ def preview_stages(
if region is None:
return None
spec = region.spec
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():
expansion = expand_by_red_boundary(
region.spec,
region.cell_mask,
contour_features,
route_line,
region.split.upstream,
floor,
)
if expansion is None:
logger.warning("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.")
return StagePreview(region=region)
# 이후 단계(유역 제원)가 표고를 쓰므로 밴드 표고를 지형 격자로 함께 들고 간다.
terrain = TerrainGrid(
spec=spec,
elevation=descent.band_elevation,
valid=descent.valid,
receiver=descent.receiver,
step_length=descent.step_length,
analysis = expansion.analysis
logger.info(
"배수유역: 단계 분석 %.1fs (확장 %d회, 최종 셀 %d개)",
time.perf_counter() - started,
expansion.rounds,
analysis.spec.size,
)
return StagePreview(
region=region,
spec=analysis.spec,
domain=analysis.domain,
terrain=analysis.terrain,
road=analysis.road,
flow=analysis.flow,
descent=analysis.descent,
expand_rounds=expansion.rounds,
expand_closed=expansion.closed,
expand_added_cells=expansion.added_cells,
)
road = rasterize_road(spec, route_line, DRAINAGE_ROAD_WIDTH_M)
# 확정된 상류 세류망을 따라 흐름을 새긴다 — 세류선 위 셀과 그리로 흘러드는 셀은
# 반드시 도로에 도달해야 한다.
terrain, burned = burn_stream_flow(terrain, road, region.split.upstream)
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, descent=descent)
# ── ③~④ 격자 해석 (캐시 대상) ───────────────────────────────────────────────
@@ -0,0 +1,221 @@
"""해석 영역 확장 — 최외곽의 **적색 셀** 주변으로 넓히며 다시 분석한다.
적색 셀이 해석 영역 최외곽에 있다는 것은 그 바깥에서 물이 더 흘러 들어온다는 뜻이다.
거기서 멈추면 유역이 잘린다. 반대로 최외곽이 전부 파랑이면 그 바깥 물은 도로로 오지
않으므로 더 볼 필요가 없다.
① 현재 해석 영역의 최외곽 셀 중 **적색**인 것을 찾는다
② 그 주변으로 한 겹(설정 폭) 넓힌다
③ 넓힌 영역으로 흐름 방향·색을 다시 분석한다
④ **새로 추가된 셀에 적색이 하나도 없으면 종료** (2026-07-31 사용자 지시)
격자 bbox에 닿으면 격자 자체도 셀 정수배로 넓힌다 — 도로 시작점 기준 격자점은 유지된다.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
import numpy as np
from shapely.geometry import LineString
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,
burn_stream_flow,
classify_flow,
outermost_cells,
rasterize_road,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import GridSpec, TerrainGrid
from config.config_system import (
DRAINAGE_RED_EXPAND_BAND_M,
DRAINAGE_RED_EXPAND_MAX_ROUNDS,
DRAINAGE_ROAD_WIDTH_M,
)
logger = logging.getLogger(__name__)
@dataclass
class GridAnalysis:
"""한 회차 분석 결과 — 격자·해석 영역·방향·색까지 한 묶음."""
spec: GridSpec
domain: np.ndarray # (R, C) bool — 해석 대상 셀
descent: ContourDescent
terrain: TerrainGrid
road: RoadRaster
flow: FlowClassification
@dataclass
class RedExpansion:
"""확장 루프 결과."""
analysis: GridAnalysis
rounds: int # 실제로 넓힌 횟수 (0 = 처음부터 최외곽에 적색이 없었음)
closed: bool # 새로 추가한 셀에 적색이 없어 스스로 멈췄는가
added_cells: int # 확장으로 늘어난 셀 수
def analyze_domain(
spec: GridSpec,
domain: np.ndarray,
contour_features: list[dict[str, Any]],
route_line: LineString,
upstream_streams: list[LineString],
elevation_floor_m: float | None = None,
) -> GridAnalysis | None:
"""주어진 격자·해석 영역에 대해 등고선 하강 방향과 도로 도달 색을 한 번 계산한다."""
descent = build_contour_descent(spec, contour_features, domain, elevation_floor_m)
if not descent.valid.any():
return None
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)
terrain, burned = burn_stream_flow(terrain, road, upstream_streams)
flow = classify_flow(terrain, road, burned, azimuth=descent.azimuth)
return GridAnalysis(
spec=spec, domain=domain, descent=descent, terrain=terrain, road=road, flow=flow
)
def expand_by_red_boundary(
spec: GridSpec,
domain: np.ndarray,
contour_features: list[dict[str, Any]],
route_line: LineString,
upstream_streams: list[LineString],
elevation_floor_m: float | None = None,
band_m: float = DRAINAGE_RED_EXPAND_BAND_M,
max_rounds: int = DRAINAGE_RED_EXPAND_MAX_ROUNDS,
) -> RedExpansion | None:
"""최외곽 적색 셀 주변으로 넓히며, 새로 추가한 셀에 적색이 없을 때까지 반복한다."""
analysis = analyze_domain(
spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m
)
if analysis is None:
return None
band_cells = max(1, int(round(band_m / spec.cell_m)))
started_cells = int(domain.sum())
rounds = 0
closed = False
for attempt in range(max_rounds):
current = analysis.spec
reaches = analysis.flow.reaches_road.reshape(current.n_rows, current.n_cols)
rim_red = outermost_cells(analysis.domain) & reaches
if not rim_red.any():
closed = True # 최외곽이 전부 파랑 — 바깥 물은 도로로 오지 않는다
break
grown_spec, grown_domain, grown_rim = _grow_for_rim(
current, analysis.domain, rim_red, band_cells
)
widened = grown_domain | _dilate_by(grown_rim, band_cells)
added_mask = widened & ~grown_domain
if not added_mask.any():
closed = True
break
logger.info(
"배수유역: %d회차 확장 — 최외곽 적색 %d셀 주변 %.0fm, 셀 %d개 추가",
attempt + 1,
int(rim_red.sum()),
band_m,
int(added_mask.sum()),
)
widened_analysis = analyze_domain(
grown_spec, widened, contour_features, route_line, upstream_streams, elevation_floor_m
)
if widened_analysis is None:
break
analysis = widened_analysis
rounds += 1
added_reaches = widened_analysis.flow.reaches_road.reshape(
grown_spec.n_rows, grown_spec.n_cols
)
if not (added_mask & added_reaches).any():
closed = True # 새로 추가한 셀에 적색이 없다 — 여기까지가 유역이다
logger.info("배수유역: 새로 추가한 셀에 적색이 없어 확장을 멈춥니다.")
break
else:
logger.warning("배수유역: 확장 상한(%d회)에 도달했습니다.", max_rounds)
added = int(analysis.domain.sum()) - started_cells
logger.info(
"배수유역: 확장 %d회, 셀 %d%d (+%d), %s",
rounds,
started_cells,
int(analysis.domain.sum()),
added,
"닫힘" if closed else "미닫힘",
)
return RedExpansion(analysis=analysis, rounds=rounds, closed=closed, added_cells=added)
def _dilate_by(mask: np.ndarray, steps: int) -> np.ndarray:
"""8이웃 팽창을 `steps`회 반복한다(정사각 커널이라 반경 = steps 셀)."""
rows, cols = mask.shape
result = mask
for _ in range(steps):
padded = np.zeros((rows + 2, cols + 2), dtype=bool)
padded[1:-1, 1:-1] = result
grown = np.zeros_like(result)
for row_shift in (0, 1, 2):
for col_shift in (0, 1, 2):
grown |= padded[row_shift : row_shift + rows, col_shift : col_shift + cols]
result = grown
return result
def _grow_for_rim(
spec: GridSpec, domain: np.ndarray, rim: np.ndarray, band_cells: int
) -> tuple[GridSpec, np.ndarray, np.ndarray]:
"""적색 최외곽이 격자 bbox에 닿았으면 그 방향으로 격자를 넓히고 마스크를 옮겨 담는다.
격자는 셀 정수배로만 넓히므로 도로 시작점 기준 격자점이 그대로 유지된다.
"""
north = band_cells if rim[0, :].any() else 0
south = band_cells if rim[-1, :].any() else 0
west = band_cells if rim[:, 0].any() else 0
east = band_cells if rim[:, -1].any() else 0
if not (north or south or west or east):
return spec, domain, rim
grown = GridSpec(
x_min=spec.x_min - west * spec.cell_m,
y_max=spec.y_max + north * spec.cell_m,
cell_m=spec.cell_m,
n_rows=spec.n_rows + north + south,
n_cols=spec.n_cols + west + east,
)
new_domain = np.zeros((grown.n_rows, grown.n_cols), dtype=bool)
new_rim = np.zeros_like(new_domain)
new_domain[north : north + spec.n_rows, west : west + spec.n_cols] = domain
new_rim[north : north + spec.n_rows, west : west + spec.n_cols] = rim
logger.info(
"배수유역: 격자 확대 %d×%d%d×%d (북%d%d%d%d 셀)",
spec.n_rows,
spec.n_cols,
grown.n_rows,
grown.n_cols,
north,
south,
west,
east,
)
return grown, new_domain, new_rim
+28 -14
View File
@@ -216,7 +216,9 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
)
region = preview.region
to_lonlat = prepared["to_lonlat"]
spec = region.spec
# 확장을 거치면 격자·해석 영역이 1차 영역보다 커진다. 최종본을 써야 화면과 어긋나지 않는다.
spec = preview.spec or region.spec
domain = preview.domain if preview.domain is not None else region.cell_mask
payload = {
"status": "success",
"project_id": str(project_id),
@@ -235,21 +237,28 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
"cell_m": spec.cell_m,
"rows": spec.n_rows,
"cols": spec.n_cols,
# bbox 전체 셀 수와, 1차 영역에 걸쳐 실제로 생성된 셀 수.
# bbox 전체 셀 수와, 해석 영역에 실제로 생성된 셀 수(확장 반영).
"bbox_cells": spec.size,
"cells": region.active_cells,
"cells": int(domain.sum()) if domain is not None else 0,
"width_m": round(spec.n_cols * spec.cell_m, 1),
"height_m": round(spec.n_rows * spec.cell_m, 1),
# 격자 bbox 링. 프론트는 이 사각형을 rows×cols로 나눠 행·열 좌표를 얻는다.
"bbox_lonlat": _grid_bbox_lonlat(spec, to_lonlat),
# 실제 생성된 셀을 행별 연속 구간 [행, 시작열, 끝열]으로 압축해 보낸다.
# 셀을 낱개로 보내면 수십만 건이라 응답이 감당되지 않는다.
"row_spans": [list(span) for span in mask_row_spans(region.cell_mask)]
if region.cell_mask is not None
"row_spans": [list(span) for span in mask_row_spans(domain)]
if domain is not None
else [],
},
# 최외곽 적색 셀 주변 확장 결과.
"expansion": {
"rounds": preview.expand_rounds,
"closed": preview.expand_closed,
"added_cells": preview.expand_added_cells,
"initial_cells": region.active_cells,
},
# 셀별 흐름 방향과 도로 도달 여부. row_spans 순서(행 → 구간 → 열 오름차순)로 1바이트씩.
"flow": _flow_payload(preview, region),
"flow": _flow_payload(preview, domain),
}
# 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
payload["saved_to"] = write_stage(
@@ -275,19 +284,24 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
},
to_lonlat,
)
_write_stage_arrays(prepared["stored_path"], preview, region, spec)
_write_stage_arrays(prepared["stored_path"], preview, domain, spec)
return payload
def _write_stage_arrays(stored_path: str, preview: Any, region: Any, spec: Any) -> None:
def _write_stage_arrays(stored_path: str, preview: Any, domain: Any, spec: Any) -> None:
"""격자 규모 배열(셀 마스크·흐름 방향·도달 여부)을 단계별 `.npz`로 남긴다."""
if region.cell_mask is not None:
if domain is not None:
write_grid_arrays(
stored_path,
"primary_region",
spec,
{"mask": region.cell_mask},
{"cells": region.active_cells, "bbox_cells": spec.size},
{"mask": domain},
{
"cells": int(domain.sum()),
"bbox_cells": spec.size,
"expand_rounds": preview.expand_rounds,
"expand_closed": preview.expand_closed,
},
)
flow = preview.flow
if flow is None:
@@ -322,7 +336,7 @@ def _write_stage_arrays(stored_path: str, preview: Any, region: Any, spec: Any)
)
def _flow_payload(preview: Any, region: Any) -> dict[str, Any] | None:
def _flow_payload(preview: Any, domain: Any) -> dict[str, Any] | None:
"""셀별 흐름 방향·도로 도달 여부를 바이트 배열로 압축한다.
셀이 수십만 개라 JSON 객체로는 못 보낸다. 셀 하나당 1바이트로 줄이고 base64로 싣는다:
@@ -331,9 +345,9 @@ def _flow_payload(preview: Any, region: Any) -> dict[str, Any] | None:
바이트 순서는 `grid.row_spans`를 행 → 구간 → 열 오름차순으로 훑은 순서와 같다.
"""
flow = preview.flow
if flow is None or region.cell_mask is None:
if flow is None or domain is None:
return None
order = np.flatnonzero(region.cell_mask.reshape(-1))
order = np.flatnonzero(domain.reshape(-1))
analyzed = flow.analyzed[order]
reaches = flow.reaches_road[order]
packed = np.clip(flow.direction[order], 0, AZIMUTH_INVALID).astype(np.uint8)
@@ -638,10 +638,15 @@ export function createDrainagePanel(): DrainagePanel {
`최외곽 출발 ${region.flow.outer_seeds.toLocaleString()} + ` +
`내부 보충 ${region.flow.interior_seeds.toLocaleString()}${burned}`
: " · 흐름 판정 없음";
const expansion = region.expansion
? ` · 확장 ${region.expansion.rounds}` +
`(${region.expansion.initial_cells.toLocaleString()}${cells}셀, ` +
`${region.expansion.closed ? "닫힘" : "상한 도달"})`
: "";
return (
`1차 영역(반경 ${region.radius_m}m): 상류망 ${region.upstream_lines.length}조각 채택 / ` +
`하류망 ${region.downstream_lines.length}조각·미연결 ${region.no_contact_count}개 제외 · ` +
`격자 ${region.grid.cell_m}m(도로 시점 기준) 셀 ${cells}${outside}${flow}`
`격자 ${region.grid.cell_m}m(도로 시점 기준) 셀 ${cells}${outside}${expansion}${flow}`
);
}
+4
View File
@@ -251,6 +251,10 @@ DRAINAGE_INITIAL_RADIUS_M = float(os.getenv("DRAINAGE_INITIAL_RADIUS_M", "100.0"
DRAINAGE_EXPAND_STEP_M = float(os.getenv("DRAINAGE_EXPAND_STEP_M", "200.0"))
# 확장 반복 상한. 경계 링이 전부 비활성이 되면 그 전에 스스로 멈춘다(안전핀).
DRAINAGE_MAX_EXPAND_ROUNDS = int(os.getenv("DRAINAGE_MAX_EXPAND_ROUNDS", "6"))
# 최외곽 적색 셀 주변을 한 회차에 넓히는 폭(m). 좁을수록 유역 경계가 정밀하나 회차가 늘어난다.
DRAINAGE_RED_EXPAND_BAND_M = float(os.getenv("DRAINAGE_RED_EXPAND_BAND_M", "50.0"))
# 적색 확장 반복 상한. 새로 추가한 셀에 적색이 없으면 그 전에 스스로 멈춘다(안전핀).
DRAINAGE_RED_EXPAND_MAX_ROUNDS = int(os.getenv("DRAINAGE_RED_EXPAND_MAX_ROUNDS", "20"))
# 격자 셀 수 권장 상한. 넘으면 **경고만** 남기고 그대로 계산한다 — 격자 크기 자동 조절은
# 하지 않는다(2026-07-31 사용자 지시). 느리면 위 DRAINAGE_GRID_SIZE_M을 직접 올린다.
DRAINAGE_MAX_GRID_CELLS = int(os.getenv("DRAINAGE_MAX_GRID_CELLS", "16000000"))