diff --git a/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py b/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py index ad1dce02b..337223f36 100644 --- a/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py +++ b/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py @@ -18,21 +18,31 @@ from __future__ import annotations import hashlib import json import logging +import math +import multiprocessing +from concurrent.futures import ProcessPoolExecutor, as_completed from pathlib import Path from typing import Any, Callable +import numpy as np +import shapely +from shapely.geometry import LineString, MultiLineString + from B05_Profile.B05_Profile_Engine_Grade import ( legal_grade_limit_pct, legal_plan_radius_min_m, resolve_design_speed, ) from B05_Profile.B05_Profile_Engine_RouteInitial import ( + _POOLS, + _TERRAINS, generate_initial_route, load_terrain_and_streams, ) from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint from common_util.common_util_json import atomic_write_json from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER +from config.config_system_terrain import PLAN_ANALYZE_WORKERS from D03_PlanAnalysis.D03_PlanAnalysis_Terrain import PLAN_CORRIDOR_M, models_dir_of logger = logging.getLogger(__name__) @@ -43,6 +53,9 @@ SELECTION = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": # D 등록에 지형 구분 칸 없음 — B `_road_settings` 못 읽을 때 폴백 그대로(가장 완화) TERRAIN_TYPE = "special" ALGO = 1 +# 계류 거르기 칸 · 지형 상자 여유(m) — `near_streams` +STREAM_TILE_M = 200.0 +STREAM_PAD_M = 100.0 Progress = Callable[[int, str], None] Loader = Callable[..., tuple[Any, list]] @@ -90,6 +103,38 @@ def _route_view(result: dict[str, Any]) -> dict[str, Any]: return {key: result.get(key) for key in keep} +def _engine_inputs(criteria: dict[str, Any]) -> dict[str, Any]: + return { + "weights": None, + "max_grade": criteria["max_grade"], + "min_radius_m": criteria["min_radius_m"], + "stream_offset_m": None, + "grade_smooth_m": None, + "target_grade": None, + "with_profile": False, + "design_speed": criteria["design_speed"], + } + + +def segment_fingerprint( + seg: dict[str, Any], + terrain: Any, + streams: list, + criteria: dict[str, Any], + corridor_m: float = PLAN_CORRIDOR_M, +) -> str: + """구간 입력 지문 — 갈래 넷 B 저장 키 · 복도 · 판.""" + expected = [_point(seg["start"]), _point(seg["end"])] + inputs = _engine_inputs(criteria) + keys = [ + request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0] + for mode in PLAN_MODES + ] + return hashlib.blake2b( + repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16 + ).hexdigest() + + def analyze_segment( project_root: Path, seg: dict[str, Any], @@ -100,23 +145,8 @@ def analyze_segment( ) -> dict[str, Any]: """연도 구간 하나 · 갈래 넷 — 입력 지문이 저장본과 같으면 그대로(`stored` True).""" expected = [_point(seg["start"]), _point(seg["end"])] - inputs = { - "weights": None, - "max_grade": criteria["max_grade"], - "min_radius_m": criteria["min_radius_m"], - "stream_offset_m": None, - "grade_smooth_m": None, - "target_grade": None, - "with_profile": False, - "design_speed": criteria["design_speed"], - } - keys = [ - request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0] - for mode in PLAN_MODES - ] - fingerprint = hashlib.blake2b( - repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16 - ).hexdigest() + inputs = _engine_inputs(criteria) + fingerprint = segment_fingerprint(seg, terrain, streams, criteria, corridor_m) stored = read_segment(project_root, seg) if stored is not None and stored.get("fingerprint") == fingerprint: return {**stored, "stored": True} @@ -148,6 +178,60 @@ def analyze_segment( return {**result, "stored": False} +def near_streams(terrain: Any, streams: list) -> list: + """지형 상자 안 어느 점에서든 가장 가까운 계류일 수 있는 선만 — 나머지는 엔진 값에 안 닿음. + + 칸마다 닿는 거리 = 칸 가운데 계류 거리 + 칸 반대각(거리는 1-립시츠) — 그 안에 드는 선만 남기면 + 상자 안 점의 계류 거리 · 건너기 · 격자 거리가 통째와 같음(통째 거리 = 추천 시간의 큰 몫). + """ + if not streams: + return streams + x0, x1 = float(np.min(terrain.x)) - STREAM_PAD_M, float(np.max(terrain.x)) + STREAM_PAD_M + y0, y1 = float(np.min(terrain.y)) - STREAM_PAD_M, float(np.max(terrain.y)) + STREAM_PAD_M + xs = np.linspace(x0, x1, max(1, math.ceil((x1 - x0) / STREAM_TILE_M)) + 1) + ys = np.linspace(y0, y1, max(1, math.ceil((y1 - y0) / STREAM_TILE_M)) + 1) + lo_x, lo_y = np.meshgrid(xs[:-1], ys[:-1]) + hi_x, hi_y = np.meshgrid(xs[1:], ys[1:]) + tiles = shapely.box(lo_x.ravel(), lo_y.ravel(), hi_x.ravel(), hi_y.ravel()) + reach = shapely.distance(shapely.centroid(tiles), MultiLineString(streams)) + 0.5 * math.hypot( + xs[1] - xs[0], ys[1] - ys[0] + ) + lines = np.array([LineString(line) for line in streams], dtype=object) + keep = (shapely.distance(lines[:, None], tiles[None, :]) <= reach[None, :]).any(axis=1) + return [line for line, kept in zip(streams, keep) if kept] + + +def _load(project_root: Path, year: int, loader: Loader) -> tuple[Any, list]: + terrain, streams = loader( + project_root, + SELECTION, + models_dir_of(project_root, year), + derived_dir(project_root) / f"cost_{year}", + ) + return terrain, near_streams(terrain, streams) + + +def _segment_job( + project_root: Path, + seg: dict[str, Any], + criteria: dict[str, Any], + corridor_m: float, + loader: Loader, +) -> dict[str, Any]: + """연도 구간 하나 — 지형 읽기부터(제 프로세스면 끝날 때 비용면 · 묶음 메모리를 놓음).""" + terrain, streams = _load(project_root, int(seg["year"]), loader) + return analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m) + + +def _worker_job(*args: Any) -> dict[str, Any]: + """나란히 셈 프로세스 — 구간 하나 뒤 B 기억(비용면 · 묶음)을 비워 다음 구간 메모리를 놓음.""" + try: + return _segment_job(*args) + finally: + _POOLS.clear() + _TERRAINS.clear() + + def analyze_plan( project_root: Path, segments: list[dict[str, Any]], @@ -156,23 +240,52 @@ def analyze_plan( loader: Loader = load_terrain_and_streams, corridor_m: float = PLAN_CORRIDOR_M, on_progress: Progress | None = None, + workers: int = PLAN_ANALYZE_WORKERS, ) -> list[dict[str, Any]]: - """연도 구간마다 지형(연도 DEM · 연도 비용면 캐시)을 읽어 `analyze_segment`.""" - results = [] - for i, seg in enumerate(segments): - year = int(seg["year"]) + """연도 구간마다(연도 DEM · 연도 비용면 캐시) `analyze_segment`. + + 다시 셀 구간이 둘 이상이면 프로세스 `workers` 개로 나란히 — 구간끼리 입력 · 저장본이 따로라 + 결과 같음. `workers` 1 = 이 프로세스(가짜 지형 시험).""" + + def report(percent: int, message: str) -> None: if on_progress is not None: - on_progress(100 * i // len(segments), f"{year}년 구간 노선 추천 중") - terrain, streams = loader( - project_root, - SELECTION, - models_dir_of(project_root, year), - derived_dir(project_root) / f"cost_{year}", - ) - results.append(analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m)) - if on_progress is not None: - on_progress(100, "분석 끝") - return results + on_progress(percent, message) + + results: list[dict[str, Any] | None] = [None] * len(segments) + todo = [] + for i, seg in enumerate(segments): + stored = read_segment(project_root, seg) + if stored is not None: # 저장본이 있을 때만 이 프로세스에서 지형을 읽어 지문을 봄 + terrain, streams = _load(project_root, int(seg["year"]), loader) + if stored.get("fingerprint") == segment_fingerprint( + seg, terrain, streams, criteria, corridor_m + ): + results[i] = {**stored, "stored": True} + continue + todo.append(i) + + count = min(workers, len(todo)) + if count <= 1: + for n, i in enumerate(todo): + report(100 * n // len(segments), f"{int(segments[i]['year'])}년 구간 노선 추천 중") + results[i] = _segment_job(project_root, segments[i], criteria, corridor_m, loader) + else: + years = "·".join(str(int(segments[i]["year"])) for i in todo) + report(0, f"{years}년 구간 노선 추천 중(나란히 {count})") + context = multiprocessing.get_context("spawn") + with ProcessPoolExecutor(count, mp_context=context) as pool: + futures = { + pool.submit(_worker_job, project_root, segments[i], criteria, corridor_m, loader): i + for i in todo + } + for n, future in enumerate(as_completed(futures), 1): + i = futures[future] + results[i] = future.result() + report( + 100 * n // len(todo), f"{int(segments[i]['year'])}년 구간 끝 ({n}/{len(todo)})" + ) + report(100, "분석 끝") + return results # type: ignore[return-value] def pick_mode(result: dict[str, Any], wanted: str | None) -> str | None: diff --git a/config/config_system_terrain.py b/config/config_system_terrain.py index 113e576e5..62d543a72 100644 --- a/config/config_system_terrain.py +++ b/config/config_system_terrain.py @@ -149,6 +149,9 @@ SHEET_SURFACE_METHODS = [ ).split(",") if method.strip() ] +# D 예정노선 [분석] 연도 구간을 나란히 셀 프로세스 수 상한(프로세스 하나 약 0.8 GB) — 여럿이 동시에 +# 돌리면 줄임. +PLAN_ANALYZE_WORKERS = int(os.getenv("PLAN_ANALYZE_WORKERS", "3")) # 확정에 쓸 기본 방식 — 라플라스 (2026-08-30 사용자 확정). SHEET_SURFACE_DEFAULT_METHOD = os.getenv("SHEET_SURFACE_DEFAULT_METHOD", "laplace") # 자동 전처리가 만드는 도엽 방식 — 기본 하나뿐이다. 여섯 방식을 매번 다 만들면 diff --git a/resources/tester/test_530_2_plan_analyze.py b/resources/tester/test_530_2_plan_analyze.py index 1e50b9275..cbab24f0f 100644 --- a/resources/tester/test_530_2_plan_analyze.py +++ b/resources/tester/test_530_2_plan_analyze.py @@ -17,8 +17,7 @@ SEGMENTS = [ ] -@pytest.fixture(scope="module") -def hill() -> Terrain: +def _hill() -> Terrain: x = np.arange(0.0, 1102.0, 2.0) y = np.arange(0.0, 402.0, 2.0) xx, yy = np.meshgrid(x, y) @@ -26,6 +25,16 @@ def hill() -> Terrain: return Terrain(x, y, z, np.ones(xx.shape, dtype=bool)) +def hill_loader(root, selection, models_dir, cache_dir): + """나란히 셈 프로세스가 이름으로 부르는 지형(모듈 맨 위 함수).""" + return _hill(), [] + + +@pytest.fixture(scope="module") +def hill() -> Terrain: + return _hill() + + @pytest.fixture def run(tmp_path, hill): _POOLS.clear() @@ -37,7 +46,7 @@ def run(tmp_path, hill): def go(criteria=CRITERIA, corridor_m=150.0): return analyze.analyze_plan( - tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m + tmp_path, SEGMENTS, criteria, loader=loader, corridor_m=corridor_m, workers=1 ) go.calls = calls @@ -91,6 +100,54 @@ def test_same_input_reads_store(run, monkeypatch) -> None: run(criteria={**CRITERIA, "max_grade": 0.12}) +def test_parallel_years_match_sequential(tmp_path) -> None: + """연도 구간 나란히(프로세스 둘) = 차례로 — 갈래 넷 노선 좌표 · 잰 값 차 0(531-6).""" + _POOLS.clear() + seq = analyze.analyze_plan( + tmp_path / "a", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=1 + ) + _POOLS.clear() + steps = [] + par = analyze.analyze_plan( + tmp_path / "b", + SEGMENTS, + CRITERIA, + loader=hill_loader, + corridor_m=150.0, + workers=2, + on_progress=lambda p, m: steps.append(m), + ) + assert [r["routes"] for r in par] == [r["routes"] for r in seq] + assert [r["fingerprint"] for r in par] == [r["fingerprint"] for r in seq] + assert "나란히 2" in steps[0] and steps[-1] == "분석 끝" + # 저장본은 제 프로세스에서 지문만 보고 그대로 + again = analyze.analyze_plan( + tmp_path / "b", SEGMENTS, CRITERIA, loader=hill_loader, corridor_m=150.0, workers=2 + ) + assert [r["stored"] for r in again] == [True, True] + _POOLS.clear() + + +def test_near_streams_keeps_every_nearest_line(hill) -> None: + """계류 거르기 — 지형 상자 안 점의 계류 거리 · 건너기가 통째와 같고 먼 선은 빠짐(531-6).""" + import shapely + from shapely.geometry import LineString, MultiLineString + + rng = np.random.default_rng(7) + streams = [] + for _ in range(80): # 지형(1.1 × 0.4 km) 둘레 5 km 에 흩은 계류 + start = rng.uniform(-2000.0, 3000.0, 2) + streams.append([tuple(start + step) for step in np.cumsum(rng.normal(0, 60, (6, 2)), 0)]) + kept = analyze.near_streams(hill, streams) + assert 0 < len(kept) < len(streams) + points = shapely.points(rng.uniform((0.0, 0.0), (1100.0, 400.0), (4000, 2))) + whole, part = MultiLineString(streams), MultiLineString(kept) + assert np.array_equal(shapely.distance(points, whole), shapely.distance(points, part)) + route = LineString([(0.0, 0.0), (1100.0, 400.0)]) + assert route.intersection(whole).equals(route.intersection(part)) + assert analyze.near_streams(hill, []) == [] + + def test_corridor_argument_keeps_b_default(hill) -> None: line = [SEGMENTS[0]["start"], SEGMENTS[0]["end"]] base = SearchGrid(line, hill, [], stream_offset_m=10.0)