feat(B05): 격자를 1차 영역 안에만 생성, 원점을 도로 시작점에 고정

bbox 사각형만 그리던 것을 실제 셀 생성으로 바꾼다.

- 격자 원점을 도로 시작점에 맞춘다(grid_spec_from_bounds anchor_xy).
  bbox 좌상단 기준이면 1차 영역이 조금만 달라져도 격자가 통째로 밀려
  이전 결과와 셀이 대응되지 않는다.
- build_cell_mask: 1차 영역에 조금이라도 걸치는 셀만 생성(all_touched).
  합성 검증에서 bbox 360,000셀 중 197,623셀(55%)만 생성.
- mask_row_spans: 셀을 행별 연속 구간으로 압축해 응답에 싣는다.
  낱개 대비 0.23% 크기(451구간 vs 197,623셀).
- expand_grid_spec 을 셀 정수배 확장으로 바꿔 확장해도 격자점이 유지된다.
- grid_transform 을 Flow -> Grid 로 옮겨 중복 제거.
- 프론트: bbox 전체 눈금 대신 생성된 셀만 흐린 선으로. 셀이 2px 미만이면
  구간을 통짜로 흐리게 칠하고 확대하면 셀 하나하나가 보인다.
- 저장: 01_primary_region_cells.npz 에 격자 규격 + 셀 마스크 기록
  (셀 수십만 개라 GeoJSON 폴리곤으로는 못 남긴다)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-31 18:05:58 +09:00
co-authored by Claude Fable 5
parent a77ee317a5
commit 683a5c9198
7 changed files with 204 additions and 78 deletions
+6 -1
View File
@@ -316,11 +316,16 @@ export interface DrainagePrimaryRegion {
cell_m: number;
rows: number;
cols: number;
/** bbox 전체 셀 수(참고값). */
bbox_cells: number;
/** 1차 영역에 걸쳐 실제로 생성된 셀 수. */
cells: number;
width_m: number;
height_m: number;
/** 격자 bbox 링. 화면은 여기에 rows×cols 간격으로 실제 셀을 그린다. */
/** 격자 bbox 링. 화면은 이 사각형을 rows×cols로 나눠 셀 좌표를 얻는다. */
bbox_lonlat: Array<[number, number]>;
/** 실제 생성된 셀 구간 [행, 시작열, 끝열(포함)]. 낱개 셀 대신 구간으로 온다. */
row_spans: Array<[number, number, number]>;
};
/** 영구저장소에 남긴 검증용 GeoJSON 경로. */
saved_to: string | null;
@@ -19,6 +19,7 @@ from datetime import datetime
from pathlib import Path
from typing import Any, Callable
import numpy as np
from shapely.geometry.base import BaseGeometry
from common_util.common_util_storage import resolve_stored_project_path
@@ -120,6 +121,35 @@ 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`로 남긴다.
셀이 수십만 개라 GeoJSON 폴리곤으로는 못 남긴다. 격자 원점·셀 크기와 bool 마스크만
저장하면 어느 셀이 생성됐는지 그대로 복원된다.
"""
prefix = STAGES.get(stage)
if prefix is None or mask is None:
return None
directory = drainage_dir(stored_path)
target = directory / f"{prefix}_{stage}_cells.npz"
try:
directory.mkdir(parents=True, exist_ok=True)
np.savez_compressed(
target,
x_min=spec.x_min,
y_max=spec.y_max,
cell_m=spec.cell_m,
n_rows=spec.n_rows,
n_cols=spec.n_cols,
mask=mask,
)
except OSError:
logger.warning("배수유역: 셀 마스크 저장 실패 (%s)", target)
return None
logger.info("배수유역: 셀 마스크 저장 — %s (%d셀)", target, int(mask.sum()))
return str(target)
def _update_manifest(
directory: Path, stage: str, filename: str, properties: dict[str, Any]
) -> None:
@@ -19,12 +19,15 @@ from dataclasses import dataclass
import numpy as np
from rasterio.features import rasterize, shapes
from rasterio.transform import from_origin
from scipy.spatial import cKDTree
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 GridSpec, TerrainGrid
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
GridSpec,
TerrainGrid,
grid_transform,
)
from config.config_system import (
DRAINAGE_MIN_BASIN_AREA_M2,
DRAINAGE_POLYGON_SIMPLIFY_M,
@@ -62,11 +65,6 @@ class FlowResult:
strength: np.ndarray # (K,) int64 — 도로 셀별 상류 셀 수(흐름 강도)
def grid_transform(spec: GridSpec):
"""rasterio 아핀 변환. 행 0이 북쪽(y_max)이다."""
return from_origin(spec.x_min, spec.y_max, spec.cell_m, spec.cell_m)
# ── 도로 굽기 ───────────────────────────────────────────────────────────────
@@ -23,10 +23,14 @@ from dataclasses import dataclass
from typing import Any
import numpy as np
from affine import Affine
from rasterio.features import rasterize
from rasterio.transform import from_origin
from scipy.interpolate import LinearNDInterpolator
from scipy.ndimage import distance_transform_edt
from shapely import segmentize
from shapely.geometry import LineString, shape
from shapely.geometry.base import BaseGeometry
from skimage.morphology import reconstruction
from config.config_system import (
@@ -213,41 +217,102 @@ def _outside_bounds(
def grid_spec_from_bounds(
x_min: float, y_min: float, x_max: float, y_max: float, cell_m: float
x_min: float,
y_min: float,
x_max: float,
y_max: float,
cell_m: float,
anchor_xy: tuple[float, float] | None = None,
) -> GridSpec:
"""격자 크기는 절대 자동으로 바꾸지 않는다 — config 값이 그대로 쓰인다(사용자 지시).
"""범위를 덮는 격자를 만든다. `anchor_xy`를 주면 그 점에 셀 모서리를 맞춘다.
셀 수가 많으면 경고만 남기고 그대로 진행한다. 느리면 `DRAINAGE_GRID_SIZE_M`을 사용자가
직접 올린다. 자동 강등은 도로 굽기 두께·정밀도 전제를 조용히 깨뜨려서 금지한다.
격자 원점은 **도로 시작점**에 고정한다(2026-07-31 사용자 지시). bbox 좌상단에 맞추면
1차 영역이 조금만 달라져도 격자가 통째로 밀려 이전 결과와 셀이 대응되지 않는다.
도로 시작점에 맞추면 반경·영역을 바꿔도 같은 자리의 셀은 같은 자리에 남는다.
격자 크기는 절대 자동으로 바꾸지 않는다 — config 값이 그대로 쓰인다(사용자 지시).
셀 수가 많으면 경고만 남기고 그대로 진행한다.
"""
width = max(x_max - x_min, cell_m)
height = max(y_max - y_min, cell_m)
n_cols = int(math.ceil(width / cell_m))
n_rows = int(math.ceil(height / cell_m))
if anchor_xy is not None:
anchor_x, anchor_y = anchor_xy
# 앵커에서 셀 정수배만큼 밖으로 나가 범위를 덮는다(넓어질 뿐 좁아지지 않는다).
x_min = anchor_x - math.ceil((anchor_x - x_min) / cell_m) * cell_m
y_max = anchor_y + math.ceil((y_max - anchor_y) / cell_m) * cell_m
n_cols = max(1, int(math.ceil((x_max - x_min) / cell_m)))
n_rows = max(1, int(math.ceil((y_max - y_min) / cell_m)))
if n_cols * n_rows > DRAINAGE_MAX_GRID_CELLS:
logger.warning(
"배수유역: 격자 %d×%d = %d셀 (%.0fm × %.0fm, 셀 %.1fm) — 권장 상한 %d셀 초과. "
"배수유역: 격자 %d×%d = %d셀 (%.0fm × %.0fm, 셀 %.2fm) — 권장 상한 %d셀 초과. "
"그대로 진행합니다. 느리면 DRAINAGE_GRID_SIZE_M을 올리세요.",
n_rows,
n_cols,
n_rows * n_cols,
width,
height,
n_cols * cell_m,
n_rows * cell_m,
cell_m,
DRAINAGE_MAX_GRID_CELLS,
)
return GridSpec(
x_min=x_min, y_max=y_min + n_rows * cell_m, cell_m=cell_m, n_rows=n_rows, n_cols=n_cols
)
return GridSpec(x_min=x_min, y_max=y_max, cell_m=cell_m, n_rows=n_rows, n_cols=n_cols)
def expand_grid_spec(spec: GridSpec, sides: dict[str, bool], step_m: float) -> GridSpec:
"""활성 셀이 닿은 방향으로만 격자를 넓힌다."""
x_min = spec.x_min - (step_m if sides.get("west") else 0.0)
x_max = spec.x_min + spec.n_cols * spec.cell_m + (step_m if sides.get("east") else 0.0)
y_max = spec.y_max + (step_m if sides.get("north") else 0.0)
y_min = spec.y_max - spec.n_rows * spec.cell_m - (step_m if sides.get("south") else 0.0)
return grid_spec_from_bounds(x_min, y_min, x_max, y_max, spec.cell_m)
"""활성 셀이 닿은 방향으로만 격자를 넓힌다.
셀 정수배로만 넓혀 격자 격자점(도로 시작점 기준)이 그대로 유지되게 한다.
"""
steps = max(1, int(math.ceil(step_m / spec.cell_m)))
west = steps if sides.get("west") else 0
east = steps if sides.get("east") else 0
north = steps if sides.get("north") else 0
south = steps if sides.get("south") else 0
return 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,
)
def grid_transform(spec: GridSpec) -> Affine:
"""rasterio 아핀 변환. 행 0이 북쪽(y_max)이다."""
return from_origin(spec.x_min, spec.y_max, spec.cell_m, spec.cell_m)
def build_cell_mask(spec: GridSpec, geometry: BaseGeometry) -> np.ndarray:
"""영역에 **조금이라도 걸치는** 셀만 True인 (rows, cols) 마스크.
`all_touched=True`라서 셀이 영역과 한 점만 스쳐도 생성 대상이 된다(사용자 지시).
"""
if geometry is None or geometry.is_empty:
return np.zeros((spec.n_rows, spec.n_cols), dtype=bool)
burned = rasterize(
[(geometry, 1)],
out_shape=(spec.n_rows, spec.n_cols),
transform=grid_transform(spec),
fill=0,
dtype="uint8",
all_touched=True,
)
return burned.astype(bool)
def mask_row_spans(mask: np.ndarray) -> list[tuple[int, int, int]]:
"""마스크를 행별 연속 구간 [행, 시작열, 끝열(포함)]으로 압축한다.
셀 수십만 개를 그대로 내보낼 수 없으니 구간으로 줄인다 — 프론트는 이 구간만 받아
실제 셀 사각형을 그린다.
"""
spans: list[tuple[int, int, int]] = []
for row in range(mask.shape[0]):
line = mask[row]
if not line.any():
continue
padded = np.concatenate(([False], line, [False]))
edges = np.flatnonzero(padded[1:] != padded[:-1])
for start, stop in zip(edges[0::2], edges[1::2]):
spans.append((row, int(start), int(stop) - 1))
return spans
# ── ④ TIN 보간 ──────────────────────────────────────────────────────────────
@@ -23,6 +23,7 @@ from shapely.ops import substring, unary_union
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Grid import (
ContourCloud,
GridSpec,
build_cell_mask,
grid_spec_from_bounds,
iter_linestrings,
)
@@ -240,16 +241,22 @@ def _mean_elevation(line: LineString, sampler: ElevationSampler) -> float:
@dataclass
class PrimaryRegion:
"""1차 배수유역 — 격자 범위의 근거. 검증 화면이 이 내용을 그대로 그린다."""
"""1차 배수유역과 그 안에 생성된 격자. 검증 화면이 이 내용을 그대로 그린다."""
split: StreamSplit
# 상류 세류망을 반경 버퍼해 합친 영역. 노선은 버퍼하지 않는다.
# 상류 세류망 + 노선을 반경 버퍼해 합친 영역.
area: Polygon | MultiPolygon | None
spec: GridSpec
radius_m: float
# 1차 영역에 조금이라도 걸쳐 실제로 생성된 셀 (rows, cols) bool 마스크.
cell_mask: np.ndarray | None = None
# 1차 영역 밖으로 나간 노선 길이(m). 그 구간 사면은 해석에서 빠진다는 경고 지표.
road_outside_m: float = 0.0
@property
def active_cells(self) -> int:
return 0 if self.cell_mask is None else int(self.cell_mask.sum())
def build_primary_region(
route_line: LineString,
@@ -258,11 +265,16 @@ def build_primary_region(
radius_m: float,
cell_m: float = DRAINAGE_GRID_SIZE_M,
) -> PrimaryRegion:
"""**상류 세류망 + 계획 노선**을 반경 버퍼해 합친 범위 = 1차 배수유역, bbox = 해석 격자.
"""**상류 세류망 + 계획 노선**을 반경 버퍼한 범위 = 1차 배수유역, 그 안에 격자를 생성한다.
노선 버퍼는 세류 교차가 없는 구간의 도로도 격자 안에 들어오게 한다 — 그래야 그 구간
사면이 유역으로 잡힌다. 상류 세류망 선정이 정확해진 뒤 다시 포함했다(2026-07-31 사용자 지시).
격자는 bbox를 통째로 채우지 않는다. **도로 시작점에 셀 모서리를 맞춘 뒤, 1차 영역에
조금이라도 걸치는 셀만** 생성한다(2026-07-31 사용자 지시). bbox 전체를 쓰면 영역 밖
빈 셀이 대부분이라 의미가 없고, 원점을 bbox 좌상단에 두면 영역이 조금만 변해도 격자가
통째로 밀려 이전 결과와 셀이 대응되지 않는다.
노선이 이 영역 밖으로 나가는 길이는 따로 재서 남긴다 — 그 구간은 도로 셀이 격자에
없어 유역이 잡히지 않으므로 반경을 올릴지 판단하는 근거가 된다.
"""
@@ -273,20 +285,32 @@ def build_primary_region(
logger.warning("배수유역: 상류 세류망이 없어 노선 버퍼만으로 1차 영역을 잡습니다.")
area = unary_union(geometries)
x_min, y_min, x_max, y_max = area.bounds
spec = grid_spec_from_bounds(x_min, y_min, x_max, y_max, cell_m)
road_start = route_line.coords[0]
spec = grid_spec_from_bounds(
x_min, y_min, x_max, y_max, cell_m, anchor_xy=(float(road_start[0]), float(road_start[1]))
)
cell_mask = build_cell_mask(spec, area)
outside = route_line.difference(area)
road_outside_m = float(outside.length) if not outside.is_empty else 0.0
active = int(cell_mask.sum())
logger.info(
"배수유역: 1차 영역 %.0f㎡, 범위 %.0fm × %.0fm, 격자 %d×%d (%.2fm), 노선 이탈 %.0fm/%.0fm",
"배수유역: 1차 영역 %.0f㎡ → 격자 %d×%d (%.2fm, 도로 시점 기준) 중 %d셀 생성 "
"(bbox %d셀의 %.0f%%), 노선 이탈 %.0fm/%.0fm",
area.area,
x_max - x_min,
y_max - y_min,
spec.n_rows,
spec.n_cols,
spec.cell_m,
active,
spec.size,
100.0 * active / max(spec.size, 1),
road_outside_m,
route_line.length,
)
return PrimaryRegion(
split=split, area=area, spec=spec, radius_m=radius_m, road_outside_m=road_outside_m
split=split,
area=area,
spec=spec,
radius_m=radius_m,
cell_mask=cell_mask,
road_outside_m=road_outside_m,
)
+12 -3
View File
@@ -27,7 +27,8 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Basin import (
build_drainage_watershed,
preview_primary_region,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Export import write_stage
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_Grid import mask_row_spans
from B05_wf2_Route.B05_wf2_Route_Repository import (
get_latest_route,
get_route_points,
@@ -226,11 +227,18 @@ 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,
"cells": spec.size,
# bbox 전체 셀 수와, 1차 영역에 걸쳐 실제로 생성된 셀 수.
"bbox_cells": spec.size,
"cells": region.active_cells,
"width_m": round(spec.n_cols * spec.cell_m, 1),
"height_m": round(spec.n_rows * spec.cell_m, 1),
# 격자 bbox 링(닫힌 사각형). 프론트가 여기에 cell_m 간격으로 실제 셀을 그린다.
# 격자 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
else [],
},
}
# 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
@@ -252,6 +260,7 @@ 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)
return payload
@@ -272,10 +272,11 @@ export function createDrainagePanel(): DrainagePanel {
context.stroke();
}
/** 해석 격자를 실제 셀 눈금으로 그린다.
/** 1차 영역에 걸쳐 실제로 생성된 셀만 사각형으로 그린다.
*
* 셀 간격이 화면에서 너무 촘촘하면(2px 미만) 눈금이 뭉개져 회색 덩어리가 되므로,
* 그때는 테두리만 남기고 "확대하면 셀이 보인다"는 상태를 유지한다. */
* bbox 전체를 채우지 않는다 — 백엔드가 준 행별 구간(row_spans)만 그린다.
* 셀이 화면에서 2px 미만이면 선이 뭉개져 회색 덩어리가 되므로, 그때는 구간을 통짜
* 사각형으로 채워 격자가 덮은 범위만 흐리게 보여 준다(확대하면 셀 하나하나가 보인다). */
function drawGridCells(
context: CanvasRenderingContext2D,
map: Normalizer,
@@ -286,49 +287,42 @@ export function createDrainagePanel(): DrainagePanel {
if (ring.length < 4) return;
const lons = ring.map(([lon]) => lon);
const lats = ring.map(([, lat]) => lat);
const lonMin = Math.min(...lons);
const lonMax = Math.max(...lons);
const latMin = Math.min(...lats);
const latMax = Math.max(...lats);
const ax = view.mapRect.width * view.scale;
const bx = view.mapRect.x * view.scale + (view.width / 2) * (1 - view.scale) + view.offsetX;
const ay = view.mapRect.height * view.scale;
const by = view.mapRect.y * view.scale + (view.height / 2) * (1 - view.scale) + view.offsetY;
const toX = (lon: number): number => ((lon - map.lonMin) / map.lonRange) * ax + bx;
const toY = (lat: number): number => (1 - (lat - map.latMin) / map.latRange) * ay + by;
const left = ((Math.min(...lons) - map.lonMin) / map.lonRange) * ax + bx;
const right = ((Math.max(...lons) - map.lonMin) / map.lonRange) * ax + bx;
const top = (1 - (Math.max(...lats) - map.latMin) / map.latRange) * ay + by;
const bottom = (1 - (Math.min(...lats) - map.latMin) / map.latRange) * ay + by;
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 left = toX(lonMin);
const right = toX(lonMax);
const top = toY(latMax);
const bottom = toY(latMin);
context.save();
context.setLineDash([]);
// 셀 눈금 — 열/행 수로 나눠 실제 셀 경계를 그대로 찍는다.
const cellWidthPx = Math.abs(right - left) / Math.max(region.grid.cols, 1);
const cellHeightPx = Math.abs(bottom - top) / Math.max(region.grid.rows, 1);
if (Math.min(cellWidthPx, cellHeightPx) >= 2) {
context.lineWidth = 0.5;
context.strokeStyle = "rgba(120, 113, 108, 0.35)";
context.beginPath();
for (let col = 0; col <= region.grid.cols; col += 1) {
const x = left + (right - left) * (col / region.grid.cols);
if (x < -50 || x > view.width + 50) continue;
context.moveTo(x, top);
context.lineTo(x, bottom);
context.strokeStyle = "rgba(120, 113, 108, 0.3)";
context.fillStyle = "rgba(120, 113, 108, 0.12)";
context.lineWidth = 0.5;
if (fine) context.beginPath();
spans.forEach(([row, colStart, colEnd]) => {
const y = top + cellH * row;
if (y + cellH < -50 || y > view.height + 50) 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);
return;
}
for (let row = 0; row <= region.grid.rows; row += 1) {
const y = top + (bottom - top) * (row / region.grid.rows);
if (y < -50 || y > view.height + 50) continue;
context.moveTo(left, y);
context.lineTo(right, y);
for (let col = colStart; col <= colEnd; col += 1) {
context.rect(left + cellW * col, y, cellW, cellH);
}
context.stroke();
}
// 격자 전체 테두리는 항상 그린다.
context.setLineDash([10, 6]);
context.lineWidth = 1.5;
context.strokeStyle = "rgba(120, 113, 108, 0.9)";
context.strokeRect(left, top, right - left, bottom - top);
});
if (fine) context.stroke();
context.restore();
}
@@ -522,8 +516,9 @@ export function createDrainagePanel(): DrainagePanel {
return (
`1차 영역(반경 ${region.radius_m}m): 상류망 ${region.upstream_lines.length}조각 채택 / ` +
`하류망 ${region.downstream_lines.length}조각·미연결 ${region.no_contact_count}개 제외 · ` +
`격자 ${region.grid.width_m}×${region.grid.height_m}m, ` +
`${region.grid.cell_m}m 셀 ${cells}개${outside}`
`격자 ${region.grid.cell_m}m(도로 시점 기준) 셀 ${cells}개 ` +
`/ bbox ${region.grid.width_m}×${region.grid.height_m}m ` +
`${region.grid.bbox_cells.toLocaleString()}셀${outside}`
);
}