From 06f2328b55300dfe6176892bdc91b4275b035199 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 31 Jul 2026 18:23:35 +0900 Subject: [PATCH] =?UTF-8?q?feat(B05):=20=EC=85=80=EB=B3=84=20=ED=9D=90?= =?UTF-8?q?=EB=A6=84=20=EB=B0=A9=ED=96=A5=20=ED=8C=90=EC=A0=95=20+=20?= =?UTF-8?q?=ED=99=94=EC=82=B4=ED=91=9C/=EC=A0=81=EC=B2=AD=20=ED=91=9C?= =?UTF-8?q?=EA=B8=B0=20(=ED=99=95=EC=9E=A5=20=EB=AF=B8=ED=8F=AC=ED=95=A8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 최외곽 셀부터 물길을 따라가며 도로 도달 여부를 판정하고 화면에 셀 단위로 표기한다. 격자 확장은 다음 검증 단계라 넣지 않았다. - 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 --- B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts | 14 ++ .../B05_wf2_Route_Engine_Watershed_Basin.py | 56 +++++++ .../B05_wf2_Route_Engine_Watershed_Export.py | 22 ++- .../B05_wf2_Route_Engine_Watershed_Flow.py | 122 ++++++++++++++ .../B05_wf2_Route_Engine_Watershed_Grid.py | 37 ++++- .../B05_wf2_Route_Router_Drainage.py | 79 ++++++++- .../B05_wf2_Route_UI_Drainage_Panel.ts | 157 +++++++++++++++--- 7 files changed, 444 insertions(+), 43 deletions(-) diff --git a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts index c9811d5c..fd71c833 100644 --- a/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts +++ b/B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts @@ -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; } diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py index 452ac168..c15e9db1 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Basin.py @@ -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) + + # ── ③~④ 격자 해석 (캐시 대상) ─────────────────────────────────────────────── diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py index 871ddc41..1b8ac236 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py @@ -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) diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py index 015aa261..7edd0366 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py @@ -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 { diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py index e384ae65..b4e72984 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py @@ -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] diff --git a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py index 13675fa3..fb316b9c 100644 --- a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py +++ b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py @@ -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 [] diff --git a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts index 62466fa4..b5afcb29 100644 --- a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts +++ b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts @@ -51,6 +51,18 @@ const LAYER_LABELS: Record = { 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}` ); }