"""배수유역 분석 **산출물 기록·응답 기하** 헬퍼. `B04_PreProcess_Router_Watershed` 가 700줄을 넘겨 떼어낸 조각이다(2026-09-04). B05 가 읽을 `.npz`·스테이지 기하를 남기는 쓰기 함수와, 응답용 좌표 변환 헬퍼만 모았다. 함수 본문·값은 옮기기 전 그대로이고, 라우터가 이 모듈에서 가져다 쓴다. 분석 응답 · 저장 `write_analysis` 도 여기(520-2 · primary-region · 저장 노선 [업데이트] 같은 길). """ import base64 import json import logging import math from pathlib import Path from typing import Any import numpy as np from shapely.geometry import Point, Polygon, box from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import ( drainage_dir, write_grid_arrays, write_stage, ) from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import ( AZIMUTH_INVALID, AZIMUTH_SINK, AZIMUTH_STEPS, mask_row_spans, ) from common_util.common_util_drainage_pipes import pipe_signature, route_signature from common_util.common_util_route_geometry import StructureCandidate from config.config_system import DRAINAGE_ARROW_SPACING_M, DRAINAGE_RESPONSE_FILENAME logger = logging.getLogger(__name__) # 응답 형식 판(版). 응답에 항목을 더하거나 값의 의미를 바꾸면 이 값을 올린다 — # 저장해 둔 옛 응답을 그대로 돌려주면 화면이 조용히 어긋나기 때문이다. # 2 = 강도 곡선 간격 1m(구간 합) + 유입 집중점(inflow_hotspots) 추가 (2026-08-01) # 3 = 강도 곡선 구간 기준을 round→floor로 바로잡고(2m 주기 빗살 제거), # 유입 집중점이 기본 관 옆 최소간격 안쪽을 피하도록 수정 (2026-08-01) RESPONSE_SCHEMA_VERSION = 3 def _response_path(stored_path: str) -> Path: """분석 응답 캐시 경로. 재산정하지 않는 한 이 파일을 그대로 돌려준다.""" return drainage_dir(stored_path) / DRAINAGE_RESPONSE_FILENAME def _save_response(stored_path: str, payload: dict[str, Any]) -> None: path = _response_path(stored_path) try: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as file: json.dump(payload, file, ensure_ascii=False) except OSError: logger.warning("배수유역: 분석 응답을 저장하지 못했습니다 (%s).", path) def _candidate_payload( candidate: StructureCandidate, to_lonlat: Any, ) -> dict[str, Any]: lon, lat = to_lonlat(candidate.x, candidate.y) return { "chainage_m": round(candidate.chainage_m, 2), "x": candidate.x, "y": candidate.y, "lon": lon, "lat": lat, "reason": candidate.reason, "stream_name": candidate.stream_name, } def write_analysis( project_id: Any, prepared: dict[str, Any], preview: Any, vertices: list[Any] ) -> dict[str, Any]: """분석 결과 → 응답 몸통 · 단계 산출물 · 응답 저장본(작업본 자리) — primary-region 과 저장 노선 [업데이트](520-2 · `post_drainage_live`)가 같은 길. `vertices` = 셈한 노선(지문).""" region = preview.region to_lonlat = prepared["to_lonlat"] # 확장을 거치면 격자·해석 영역이 1차 영역보다 커진다. 최종본을 써야 화면과 어긋나지 않는다. spec = preview.spec or region.spec domain = preview.domain if preview.domain is not None else region.cell_mask payload = { "status": "success", "schema_version": RESPONSE_SCHEMA_VERSION, "project_id": str(project_id), "route_source": prepared["route_source"], # 지문(342-3) — 노선 · 기본 관 자리가 지금과 다르면 이 결과는 낡음(세부 유역 안 나눔). "route_signature": route_signature(vertices), "pipe_signature": pipe_signature([pipe.chainage_m for pipe in preview.pipes]), "radius_m": region.radius_m, # 채택된 상류 세류망 = 1차 영역의 기준선. "upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream], # 도로 아래로 이어진 하류망 — 판정이 맞는지 눈으로 대조하기 위해 함께 준다. "downstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.downstream], "no_contact_count": region.split.no_contact, # 노선이 1차 영역 밖으로 나간 길이(m). 크면 반경을 올려야 한다는 신호. "road_outside_m": round(region.road_outside_m, 1), # 1차 영역(버퍼 합집합) 외곽 링 목록. "region_rings": _polygon_rings(region.area, to_lonlat), "grid": { "cell_m": spec.cell_m, "rows": spec.n_rows, "cols": spec.n_cols, # bbox 전체 셀 수와, 해석 영역에 실제로 생성된 셀 수(확장 반영). "bbox_cells": spec.size, "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(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, domain), # ⑦ 2차 전체 배수유역 외곽선(= 분수령)과 면적. "basin_polygon_lonlat": [list(to_lonlat(x, y)) for x, y in preview.basin_boundary_xy], "basin_area_m2": round(preview.basin_area_m2, 1), # ⑥ 도로 위 흐름 강도 — [누가거리 m, 그 구간으로 모이는 상류 면적 ㎡]. "strength_profile": [ [round(chainage, 1), round(area, 1)] for chainage, area in preview.strength_profile ], # ⑥-1 유입 집중점 — 기본 관으로 나눈 구역마다 물이 많이 모이는 자리. # [누가거리 m, 유입면적 ㎡, 구역 번호, 구역 내 순위]. 화면 마커·하이라이트 대상이다. "inflow_hotspots": [ [round(chainage, 1), round(area, 1), zone, rank] for chainage, area, zone, rank in preview.inflow_hotspots ], # ⑧ 기본 관 매설 위치 — 도로 × 세류선 교차점. "pipes": [_candidate_payload(pipe, to_lonlat) for pipe in preview.pipes], # B05용 평균 흐름 화살표 — [lon, lat, 방위(도), 도로도달, 셀 수]. # 세류·도로 셀을 뺀 블록 평균이라 사면 경향만 남는다. "flow_arrows": [ [*to_lonlat(x, y), round(math.degrees(angle), 1), reaches, cells] for x, y, angle, reaches, cells in preview.flow_arrows ], # 화살표 간격(m). 화면이 화살표 크기를 정할 때 쓴다 — 서로 닿지 않게 이 값보다 짧게 그린다. "arrow_spacing_m": DRAINAGE_ARROW_SPACING_M, "from_cache": False, } # 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다. payload["saved_to"] = write_stage( prepared["stored_path"], "primary_region", { "primary_region": _as_polygons(region.area), "upstream": region.split.upstream, "downstream": region.split.downstream, "route": [prepared["route_line"]], "grid_bbox": [_grid_bbox_polygon(spec)], # ⑦ 2차 전체 배수유역 외곽선, ⑧ 기본 관 위치(누가거리·근거 포함). "basin_boundary": _boundary_geometry(preview.basin_boundary_xy), "pipe": [ ( Point(pipe.x, pipe.y), { "chainage_m": round(pipe.chainage_m, 2), "reason": pipe.reason, "stream_name": pipe.stream_name, }, ) for pipe in preview.pipes ], }, { "radius_m": region.radius_m, "road_outside_m": payload["road_outside_m"], "no_contact_count": region.split.no_contact, "basin_area_m2": payload["basin_area_m2"], "pipe_count": len(preview.pipes), # 기하(bbox·셀 구간)는 파일 본문에 있으므로 요약 수치만 남긴다. "grid": { key: value for key, value in payload["grid"].items() if key not in {"bbox_lonlat", "row_spans"} }, }, to_lonlat, ) _write_stage_arrays(prepared["stored_path"], preview, domain, spec) _write_road_routing(prepared["stored_path"], preview, spec, prepared["route_line"], to_lonlat) # 응답 자체를 캐시로 남긴다 — 다음 조회는 배열을 재조립하지 않고 이 파일을 그대로 준다. _save_response(prepared["stored_path"], payload) return payload def _write_road_routing( stored_path: str, preview: Any, spec: Any, route_line: Any, to_lonlat: Any ) -> None: """B05가 세부유역을 나눌 때 쓸 최소 산출물을 남긴다. B05는 일반 사용자용이라 가벼워야 한다. 화살표(방향 코드)·밴드 표고 같은 확인용 배열은 빼고, **셀 → 도로 셀 귀속**과 도로 셀 제원만 담는다. 여기에 표고를 함께 넣는 이유는 유역 낙차를 내려면 셀 표고가 필요해서다(2026-07-31 사용자 지시). """ routing = preview.routing road = preview.road if routing is None or road is None or road.count == 0: return write_grid_arrays( stored_path, "road_routing", spec, { "road_slot": routing.road_slot, "path_length": routing.path_length, "strength": routing.strength, "road_cell_index": road.cell_index, "road_chainage": road.chainage, "elevation": preview.terrain.elevation.reshape(-1), }, { "road_cells": road.count, "reached_cells": int((routing.road_slot >= 0).sum()), "basin_area_m2": round(preview.basin_area_m2, 1), "pipe_count": len(preview.pipes), }, ) # B05가 그대로 그릴 기하 — 계획도로선 · 기본 배관 · 2차 전체 배수유역, 이 셋뿐이다. write_stage( stored_path, "road_routing", { "route": [route_line], "basin_boundary": _boundary_geometry(preview.basin_boundary_xy), "pipe": [ ( Point(pipe.x, pipe.y), {"chainage_m": round(pipe.chainage_m, 2), "reason": pipe.reason}, ) for pipe in preview.pipes ], # 평균 흐름 화살표 — B05도 같은 그림을 그려야 하므로 여기 함께 남긴다. "flow_arrow": [ ( Point(x, y), { # B05 화면은 사업지 CRS(m)로 그리므로 미터 좌표도 함께 남긴다. "x": round(x, 2), "y": round(y, 2), "azimuth_deg": round(math.degrees(angle), 1), "reaches_road": reaches, "cells": cells, }, ) for x, y, angle, reaches, cells in preview.flow_arrows ], }, { "basin_area_m2": round(preview.basin_area_m2, 1), "pipe_count": len(preview.pipes), "route_length_m": round(route_line.length, 1), "arrow_count": len(preview.flow_arrows), "arrow_spacing_m": DRAINAGE_ARROW_SPACING_M, }, to_lonlat, ) def _write_stage_arrays(stored_path: str, preview: Any, domain: Any, spec: Any) -> None: """격자 규모 배열(셀 마스크·흐름 방향·도달 여부)을 단계별 `.npz`로 남긴다.""" if domain is not None: write_grid_arrays( stored_path, "primary_region", spec, {"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: return arrays = { "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 # ⑥ 흐름 강도 곡선 — 기하가 아니라 수치 곡선이라 GeoJSON이 아닌 여기에 함께 담는다. if preview.strength_profile: curve = np.asarray(preview.strength_profile, dtype=np.float64) arrays["strength_chainage_m"] = curve[:, 0] arrays["strength_area_m2"] = curve[:, 1] write_grid_arrays( stored_path, "flow_direction", spec, arrays, { "azimuth_steps": AZIMUTH_STEPS, "sink_code": AZIMUTH_SINK, "invalid_code": AZIMUTH_INVALID, "analyzed": int(flow.analyzed.sum()), "reaches_road": int((flow.reaches_road & flow.analyzed).sum()), "no_road": int((~flow.reaches_road & flow.analyzed).sum()), "burned": 0 if flow.burned is None else int(flow.burned.sum()), "outer_seeds": flow.outer_seeds, "interior_seeds": flow.interior_seeds, "strength_points": len(preview.strength_profile), "strength_total_m2": round(sum(area for _, area in preview.strength_profile), 1), }, ) def _flow_payload(preview: Any, domain: Any) -> dict[str, Any] | None: """셀별 흐름 방향·도로 도달 여부를 바이트 배열로 압축한다. 셀이 수십만 개라 JSON 객체로는 못 보낸다. 셀 하나당 1바이트로 줄이고 base64로 싣는다: 하위 6비트(0x3F) = 32방위 코드(0~31, 0=화면 오른쪽·시계방향), 32=제자리, 33=표고 없음 최상위 비트(0x80) = 도로 도달(적색). 꺼져 있으면 미도달(백색 화살표). 바이트 순서는 `grid.row_spans`를 행 → 구간 → 열 오름차순으로 훑은 순서와 같다. """ flow = preview.flow if flow is None or domain is None: return None 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) packed |= np.where(reaches, 0x80, 0).astype(np.uint8) burned = flow.burned return { "encoding": "base64-uint8", "azimuth_steps": AZIMUTH_STEPS, "sink_code": AZIMUTH_SINK, "invalid_code": AZIMUTH_INVALID, "cells": int(order.size), "reaches_road": int((reaches & analyzed).sum()), "no_road": int((~reaches & analyzed).sum()), # 격자에는 있으나 등고선 TIN 밖이라 표고가 없어 판정 못한 셀. "unanalyzed": int((~analyzed).sum()), # 확정된 상류 세류망을 따라 흐름을 강제로 새긴 셀 수. "burned": 0 if burned is None else int(burned[order].sum()), "outer_seeds": flow.outer_seeds, "interior_seeds": flow.interior_seeds, "data": base64.b64encode(packed.tobytes()).decode("ascii"), } def _boundary_geometry(ring: list[tuple[float, float]]) -> list[Any]: """2차 유역 외곽 링을 저장용 폴리곤으로 만든다(정점 3개 미만이면 비운다).""" return [Polygon(ring)] if len(ring) >= 4 else [] def _as_polygons(geometry: Any) -> list[Any]: if geometry is None or geometry.is_empty: return [] return list(geometry.geoms) if geometry.geom_type == "MultiPolygon" else [geometry] def _grid_bbox_polygon(spec: Any) -> Polygon: x_max = spec.x_min + spec.n_cols * spec.cell_m y_min = spec.y_max - spec.n_rows * spec.cell_m return box(spec.x_min, y_min, x_max, spec.y_max) def _line_lonlat(line: Any, to_lonlat: Any) -> list[list[float]]: return [list(to_lonlat(x, y)) for x, y in line.coords] def _polygon_rings(geometry: Any, to_lonlat: Any) -> list[list[list[float]]]: """폴리곤/멀티폴리곤의 외곽 링만 뽑아 lonlat으로 바꾼다.""" if geometry is None or geometry.is_empty: return [] parts = geometry.geoms if geometry.geom_type == "MultiPolygon" else [geometry] return [[list(to_lonlat(x, y)) for x, y in part.exterior.coords] for part in parts] def _grid_bbox_lonlat(spec: Any, to_lonlat: Any) -> list[list[float]]: x_min = spec.x_min x_max = spec.x_min + spec.n_cols * spec.cell_m y_max = spec.y_max y_min = spec.y_max - spec.n_rows * spec.cell_m corners = ((x_min, y_min), (x_min, y_max), (x_max, y_max), (x_max, y_min), (x_min, y_min)) return [list(to_lonlat(x, y)) for x, y in corners]