feat(520-2): B05 배수유역 저장 노선 [업데이트] 결과 남김 · 끊김 알림 · 도엽 메모리 보관

- B04 Router_Watershed — 분석 응답 · 단계 산출물 저장을 Output.write_analysis 로 뗌(primary-region 같은 길) · 도엽 좌표 저장본 메모리 보관(2 벌)
- B04 Router_Watershed_Live — 저장 노선 기준(초안 없음 · 지문 같음) 답이면 마지막 끊김 확인 뒤 분석을 남김 · design/save 잠금 중이면 안 남김 · 남길 때 같은 잠금 쥠 · POST /drainage/live/cancel(그 차례가 마지막일 때만 지움) · 셈 멈춤 로그
- B05 Drainage_Live — 요청 끊을 때 서버에 끊김 알림 · 저장 노선 [업데이트] 답이 저장분 낡음 걷음 · 되돌리기로 초안이 없어지면 저장분의 마지막 낡음으로 표시 바로
- 시험 test_349_drainage_live_flow 고침 — 되돌리기 직후 낡음 기대가 옛 동작(답 올 때까지 낡음) · 새 동작 = 저장분 마지막 낡음(여기선 꺼짐) + 끊김 알림 확인
- 시험 test_347_2 — 「작업본에 안 씀」 옛 설명 고침 · 남김 · 초안 안 남김 · 잠금 중 안 남김 · 끊김 알림 시험 더함 · test_347_1 메모리 보관 확인

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B04_PreProcess/B04_PreProcess_Router_Watershed_Output.py

@@ -3,16 +3,21 @@
 `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,
 )
@@ -20,8 +25,176 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
     AZIMUTH_INVALID,
     AZIMUTH_SINK,
     AZIMUTH_STEPS,
+    mask_row_spans,
 )
-from config.config_system import DRAINAGE_ARROW_SPACING_M
+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(