Files
Aislo/B05_wf2_Route/B05_wf2_Route_Engine_Disposal.py
T
eomsangdonandClaude Opus 5 6c90c26eef feat(B05): 계획 유토곡선 이관 + 사토장·토취장 4옵션 자동 선정
절·성토 균형은 종단 시공계획고로 정해지고 사토장 위치도 계획고를 다시 끌어야
정리되므로, 계획용 유토곡선과 부지 선정을 B05로 옮겼다. B06은 실측 단면적으로
낸 정식 곡선과 확정·B08 인계를 맡는다.

B05 계획 유토곡선
- _UI_Profile_MassHaul.ts: 종단면 패널 안 2차 하단 슬라이드. 펼치면 12행 도면
  테이블 자리를 곡선이 대신 차지한다. 곡선 SVG를 종단 그래프와 같은 가로
  스크롤러 안 형제로 넣고 MassHaulAxis에 LONG_PAD + originOffset을 넘겨 X축을
  종단 chainageMapper와 일치시켰다.
- computeLongitudinalMassHaul(): 횡단 설계가 없는 계획 단계용 개략 엔진. 표준횡단
  노반폭을 전 구간 공통으로 물린다. 종단 기반 토량은 국내 기준상 노선계획 개산용
  이므로(2026-08-03 조사) 표준단면까지 씌워 정밀화하지 않는다.

사토장·토취장 자동 선정
- B05_wf2_Route_Engine_Disposal.py + POST /route/disposal-sites: 노선 corridor
  DEM 격자를 훑어 후보 부지를 추리고 옵션별 점수로 정렬한다. 기준 4가지 —
  비용(최단거리+하향 운반), 지형안정성(완경사·계곡 이격), 계곡부(계곡 축 매립),
  임내 공간(라이다 수고 기반 공터). 기준값 정의처는 DISPOSAL_SITE_CRITERIA.
- _UI_Profile_Disposal.ts: 범례 줄의 [토량 분배]와 [도형 위치 초기화] 사이에
  구분기호와 함께 라디오 토글 4개. 부지 카드에 위치·용량·운반거리와 기준별 근거값,
  측점·용량 사용자 정의 편집과 복원. 후보 부족으로 남은 토량은 경고로 알린다.
- 확정 시 종단 정본의 disposal_sites에 심는다(기준 미선택이면 저장분 삭제).

B06 연동
- common_util_mass_haul_sites.ts: 저장된 부지를 정식 곡선의 잔량 위치 기준으로
  운반거리를 다시 재 표시하고 mass_haul.disposal_sites로 B08에 넘긴다.

미구현(계획서에 남김): 토취장 토질 판정(지반유형이 B06 산출물), 계곡부 암거 연장.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-03 19:06:46 +09:00

433 lines
19 KiB
Python

"""사토장·토취장 후보지 자동 선정 엔진 (B05 계획 유토곡선).
유토곡선이 낸 잉여(사토)·부족(토취) 토량을 노선 주변 어디서 처리할지 고른다.
DEM 격자에서 후보 부지를 추리고 옵션별 점수로 정렬해 상위 몇 곳을 돌려준다.
── 왜 노선 corridor 안에서만 찾는가 ─────────────────────────────────
사토·토취는 운반비가 곧 비용이라, 노선에서 멀어지면 실무적으로 선택지가 아니다.
탐색 범위를 `DISPOSAL_SITE_CRITERIA["corridor_half_width_m"]`로 묶어 계산량을 노선
연장에 비례하게 유지한다(2026-08-03 사용자 확정).
── 선정 기준 4가지 ─────────────────────────────────────────────────
실무 선정 기준(토목시공학 계열 정리, 2026-08-03 조사)을 격자 판정으로 옮겼다.
cost 운반거리 최소 + 하향 운반 우대. 트럭이 흙을 싣고 오르막을 타지 않게 한다.
stability 산사태·붕괴 위험이 없는 평탄지·완경사지. 계곡부·집수 상부는 감점.
valley 계곡부 매립 — 배관(암거) 지점과 연계해 주변부를 쌓아 올리는 실무 방식.
clearing 임내 공간 — 라이다 식생 높이가 낮은 공터·벌채지를 우선.
토취장은 파내는 쪽이라 사토장보다 경사 상한이 관대하다(`borrow_max_ground_slope`).
**토질 판정은 아직 없다** — 지반유형(토사/리핑암/발파암)은 B06 횡단 설계가 만드는 값이고
B05 계획 단계에는 노선 위 측점의 지반유형조차 확정되지 않았다. 지금은 경사만 보고 거르며,
발파암 배제는 B06 확정 지반유형이 붙은 뒤에 걸어야 한다.
기준값 정의처는 `config_system.DISPOSAL_SITE_CRITERIA` 한 곳뿐이다.
"""
from __future__ import annotations
import logging
import math
from pathlib import Path
from typing import Any
import numpy as np
from B05_wf2_Route.B05_wf2_Route_Engine_Skeleton import load_or_build_skeleton
from B05_wf2_Route.B05_wf2_Route_Engine_Solver import _load_or_build_cost_surface
from config.config_system import DISPOSAL_SITE_CRITERIA
logger = logging.getLogger(__name__)
_MODELS_SUBDIR = Path("B04_wf1_Surface") / "models"
_STRUCTURED_PATH = Path("B04_wf1_Surface") / "processed" / "structured.npz"
#: 지원하는 선정 옵션. 프론트 토글 4개와 1:1로 대응한다.
DISPOSAL_OPTIONS = ("cost", "stability", "valley", "clearing")
class DisposalSiteError(ValueError):
"""후보지 계산에 필요한 입력이 없거나 형식이 어긋날 때."""
def _route_arrays(polyline: list[list[float]]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""노선 폴리라인을 (xy, z, 누가거리) 배열로 편다."""
if len(polyline) < 2:
raise DisposalSiteError("노선 좌표가 2점 미만이라 후보지를 찾을 수 없습니다.")
points = np.asarray(polyline, dtype=float)
xy = points[:, :2]
z = points[:, 2] if points.shape[1] > 2 else np.zeros(len(points))
steps = np.linalg.norm(np.diff(xy, axis=0), axis=1)
chainage = np.concatenate([[0.0], np.cumsum(steps)])
return xy, z, chainage
def _resample_route(
xy: np.ndarray, z: np.ndarray, chainage: np.ndarray, spacing_m: float
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""노선을 일정 간격으로 다시 찍고 각 점의 진행 방향 단위벡터를 함께 낸다."""
total = float(chainage[-1])
count = max(int(total // max(spacing_m, 1.0)) + 1, 2)
targets = np.linspace(0.0, total, count)
sx = np.interp(targets, chainage, xy[:, 0])
sy = np.interp(targets, chainage, xy[:, 1])
sz = np.interp(targets, chainage, z)
samples = np.column_stack([sx, sy])
tangent = np.gradient(samples, axis=0)
norm = np.linalg.norm(tangent, axis=1, keepdims=True)
tangent = np.divide(tangent, np.where(norm > 1e-9, norm, 1.0))
return samples, sz, targets, tangent
def _valley_points(project_root: Path, filter_key: str, method: str, smooth: bool) -> np.ndarray:
"""지형 스켈레톤의 계곡 축 좌표를 한 배열로 모은다. 없으면 빈 배열."""
try:
skeleton = load_or_build_skeleton(project_root, filter_key, method, smooth)
except (FileNotFoundError, OSError, ValueError) as exc:
logger.warning("사토장 선정: 지형 스켈레톤을 열지 못했습니다 — %s", exc)
return np.empty((0, 2))
collected: list[list[float]] = []
for key in ("main_valley", "minor_valley"):
for line in skeleton.get(key) or []:
for point in line.get("polyline") or []:
if len(point) >= 2:
collected.append([float(point[0]), float(point[1])])
return np.asarray(collected, dtype=float) if collected else np.empty((0, 2))
def _canopy_grid(project_root: Path) -> tuple[dict[tuple[int, int], float], float] | None:
"""라이다 점군에서 격자별 식생 높이(지표 대비 최고점)를 만든다.
구조화 점군(`structured.npz`)의 분류값을 쓰되, 분류가 없는 자료도 있으므로
**격자 안 최고점 − 최저점**을 수고로 본다. 지표가 격자 안에서 크게 기울면
그 경사분이 수고에 섞이지만, 공터 판정은 "낮은 쪽"만 골라내면 되므로
보수적(=공터를 덜 잡는) 방향이라 문제되지 않는다.
"""
path = project_root / _STRUCTURED_PATH
if not path.is_file():
return None
try:
with np.load(path, allow_pickle=False, mmap_mode="r") as data:
xyz = np.asarray(data["xyz"], dtype=float)
except (OSError, ValueError, KeyError) as exc:
logger.warning("사토장 선정: 구조화 점군을 읽지 못했습니다 — %s", exc)
return None
if xyz.size == 0:
return None
cell = float(DISPOSAL_SITE_CRITERIA["clearing_grid_m"])
keys = np.floor(xyz[:, :2] / cell).astype(np.int64)
lowest: dict[tuple[int, int], float] = {}
highest: dict[tuple[int, int], float] = {}
for (col, row), height in zip(map(tuple, keys), xyz[:, 2]):
key = (int(col), int(row))
if key not in lowest or height < lowest[key]:
lowest[key] = float(height)
if key not in highest or height > highest[key]:
highest[key] = float(height)
return {key: highest[key] - lowest[key] for key in lowest}, cell
class _Terrain:
"""DEM 격자 접근 묶음 — 표고·경사를 좌표로 바로 묻는다."""
def __init__(self, project_root: Path, filter_key: str, method: str, smooth: bool) -> None:
x, y, z, valid, dz_dx, dz_dy, grid_res = _load_or_build_cost_surface(
project_root, project_root / _MODELS_SUBDIR, filter_key, method, smooth
)
self.x = np.asarray(x, dtype=float)
self.y = np.asarray(y, dtype=float)
self.z = np.asarray(z, dtype=float)
self.valid = np.asarray(valid, dtype=bool)
self.slope = np.hypot(np.asarray(dz_dx, dtype=float), np.asarray(dz_dy, dtype=float))
self.grid_res = float(grid_res)
def index(self, px: float, py: float) -> tuple[int, int] | None:
"""좌표를 격자 (row, col)로 옮긴다. 격자 밖이거나 무효면 None."""
if self.x.size < 2 or self.y.size < 2:
return None
col = int(round((px - self.x[0]) / (self.x[1] - self.x[0])))
row = int(round((py - self.y[0]) / (self.y[1] - self.y[0])))
if not (0 <= row < self.z.shape[0] and 0 <= col < self.z.shape[1]):
return None
return (row, col) if bool(self.valid[row, col]) else None
def patch_slope(
self, row: int, col: int, radius_cells: int, slope_limit: float
) -> tuple[float, float, int]:
"""후보 주변 격자의 (평균경사, 최대경사, **쓸 수 있는 칸 수**).
마지막 값은 경사 상한을 넘지 않는 유효 칸만 센 것이다 — 부지 용량은 반경 안
전체가 아니라 실제로 흙을 올릴 수 있는 넓이라야 후보마다 값이 갈린다.
"""
r0 = max(row - radius_cells, 0)
r1 = min(row + radius_cells + 1, self.z.shape[0])
c0 = max(col - radius_cells, 0)
c1 = min(col + radius_cells + 1, self.z.shape[1])
window = self.slope[r0:r1, c0:c1]
mask = self.valid[r0:r1, c0:c1]
if not mask.any():
return (math.inf, math.inf, 0)
values = window[mask]
usable = int((mask & (window <= slope_limit)).sum())
return (float(values.mean()), float(values.max()), usable)
def _zone_center(zones: list[dict[str, Any]]) -> float | None:
"""토량 가중 평균 위치(m). 잉여/부족이 어디에 몰려 있는지의 대표값."""
total = sum(float(zone.get("volume_m3") or 0.0) for zone in zones)
if total <= 0:
return None
weighted = sum(
float(zone.get("volume_m3") or 0.0)
* 0.5
* (float(zone.get("from_m") or 0.0) + float(zone.get("to_m") or 0.0))
for zone in zones
)
return weighted / total
def _score(
option: str,
kind: str,
*,
haul_distance_m: float,
route_length_m: float,
mean_slope: float,
drop_m: float,
valley_distance_m: float | None,
canopy_m: float | None,
) -> float:
"""0~1 점수. 옵션마다 무엇을 크게 보는지만 다르고 계산 재료는 공유한다."""
near = 1.0 - min(haul_distance_m / max(route_length_m, 1.0), 1.0)
flat = 1.0 - min(mean_slope / max(float(DISPOSAL_SITE_CRITERIA["max_ground_slope"]), 1e-6), 1.0)
# 사토장은 노면보다 낮아야 트럭이 오르막을 타지 않는다(토취장은 반대로 무관).
downhill = 1.0 if kind == "spoil" and drop_m > 0 else 0.5 if drop_m > -2.0 else 0.0
buffer_m = float(DISPOSAL_SITE_CRITERIA["valley_buffer_m"])
if valley_distance_m is None:
away_from_valley = 0.5
in_valley = 0.0
else:
away_from_valley = min(valley_distance_m / max(buffer_m, 1e-6), 1.0)
in_valley = 1.0 - away_from_valley
open_ground = (
0.5
if canopy_m is None
else 1.0
- min(canopy_m / max(float(DISPOSAL_SITE_CRITERIA["clearing_max_canopy_m"]), 1e-6), 1.0)
)
if option == "cost":
return 0.6 * near + 0.25 * downhill + 0.15 * flat
if option == "stability":
return 0.5 * flat + 0.3 * away_from_valley + 0.2 * near
if option == "valley":
return 0.5 * in_valley + 0.3 * near + 0.2 * flat
return 0.5 * open_ground + 0.3 * near + 0.2 * flat
def _build_candidates(
*,
option: str,
kind: str,
terrain: _Terrain,
samples: np.ndarray,
sample_z: np.ndarray,
chainages: np.ndarray,
tangent: np.ndarray,
valley_xy: np.ndarray,
canopy: tuple[dict[tuple[int, int], float], float] | None,
focus_m: float | None,
route_length_m: float,
) -> list[dict[str, Any]]:
"""corridor를 훑어 후보 부지를 만든다. 옵션·용도별 판정을 여기서 건다."""
criteria = DISPOSAL_SITE_CRITERIA
half_width = float(criteria["corridor_half_width_m"])
offset_step = float(criteria["offset_spacing_m"])
radius_cells = max(int(round(float(criteria["flat_radius_m"]) / terrain.grid_res)), 1)
slope_limit = float(
criteria["borrow_max_ground_slope"] if kind == "borrow" else criteria["max_ground_slope"]
)
depth_m = float(criteria["borrow_depth_m"] if kind == "borrow" else criteria["fill_depth_m"])
buffer_m = float(criteria["valley_buffer_m"])
cell_area = terrain.grid_res**2
offsets = [
sign * step
for step in np.arange(offset_step, half_width + 1e-6, offset_step)
for sign in (1.0, -1.0)
]
candidates: list[dict[str, Any]] = []
for index in range(len(samples)):
# 진행방향에 수직인 법선으로 좌우를 벌린다(좌 = +, 우 = −).
normal = np.array([-tangent[index, 1], tangent[index, 0]])
for offset in offsets:
px, py = samples[index] + normal * offset
cell = terrain.index(px, py)
if cell is None:
continue
row, col = cell
mean_slope, max_slope, usable_cells = terrain.patch_slope(
row, col, radius_cells, slope_limit
)
if usable_cells <= 0:
continue
if not math.isfinite(mean_slope) or mean_slope > slope_limit:
continue
valley_distance = (
float(np.min(np.linalg.norm(valley_xy - np.array([px, py]), axis=1)))
if valley_xy.size
else None
)
# 안정성 기준은 계곡부를 빼고, 계곡부 매립 기준은 계곡부만 남긴다.
if option == "stability" and valley_distance is not None and valley_distance < buffer_m:
continue
if option == "valley" and (valley_distance is None or valley_distance > buffer_m):
continue
canopy_m: float | None = None
if canopy is not None:
grid, cell_size = canopy
canopy_m = grid.get((int(px // cell_size), int(py // cell_size)))
if option == "clearing" and (
canopy_m is None or canopy_m > float(criteria["clearing_max_canopy_m"])
):
continue
elevation = float(terrain.z[row, col])
drop_m = float(sample_z[index]) - elevation
chainage = float(chainages[index])
haul_distance = (
abs(chainage - focus_m) + abs(offset) if focus_m is not None else abs(offset)
)
candidates.append(
{
"kind": kind,
"chainage_m": round(chainage, 2),
"offset_m": round(float(offset), 2),
"side": "left" if offset > 0 else "right",
"x": round(float(px), 3),
"y": round(float(py), 3),
"elevation_m": round(elevation, 3),
# 용량 = 경사 상한을 넘지 않는 격자 면적 × 허용 두께.
"capacity_m3": round(usable_cells * cell_area * depth_m, 1),
"haul_distance_m": round(haul_distance, 1),
"mean_slope": round(mean_slope, 4),
"max_slope": round(max_slope, 4),
"drop_m": round(drop_m, 2),
"valley_distance_m": (
None if valley_distance is None else round(valley_distance, 1)
),
"canopy_m": None if canopy_m is None else round(canopy_m, 2),
"score": round(
_score(
option,
kind,
haul_distance_m=haul_distance,
route_length_m=route_length_m,
mean_slope=mean_slope,
drop_m=drop_m,
valley_distance_m=valley_distance,
canopy_m=canopy_m,
),
4,
),
}
)
return candidates
def _pick(candidates: list[dict[str, Any]], demand_m3: float) -> list[dict[str, Any]]:
"""점수 순으로 고르되 서로 겹치지 않게 띄우고, 필요한 토량이 찰 때까지만 담는다."""
criteria = DISPOSAL_SITE_CRITERIA
spacing = float(criteria["flat_radius_m"]) * 2.0
limit = int(criteria["max_candidates"])
chosen: list[dict[str, Any]] = []
remaining = max(demand_m3, 0.0)
for entry in sorted(candidates, key=lambda item: -item["score"]):
if len(chosen) >= limit:
break
if any(
abs(entry["chainage_m"] - picked["chainage_m"]) < spacing
and entry["side"] == picked["side"]
for picked in chosen
):
continue
assigned = min(entry["capacity_m3"], remaining) if remaining > 0 else 0.0
chosen.append(
{**entry, "id": f"{entry['kind']}-{len(chosen) + 1}", "assigned_m3": round(assigned, 1)}
)
remaining -= assigned
if remaining <= 0:
break
return chosen
def compute_disposal_sites(
project_root: Path,
route_polyline: list[list[float]],
*,
option: str,
filter_key: str,
method: str,
smooth: bool,
surplus_m3: float,
shortage_m3: float,
surplus_zones: list[dict[str, Any]] | None = None,
shortage_zones: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""옵션 하나에 대한 사토장·토취장 후보지를 계산한다.
잉여가 없으면 사토장을, 부족이 없으면 토취장을 빈 목록으로 돌려준다 —
필요 없는 부지를 억지로 만들면 사용자가 잘못된 선택을 하게 된다.
"""
if option not in DISPOSAL_OPTIONS:
raise DisposalSiteError(f"알 수 없는 선정 기준입니다: {option}")
xy, z, chainage = _route_arrays(route_polyline)
samples, sample_z, chainages, tangent = _resample_route(
xy, z, chainage, float(DISPOSAL_SITE_CRITERIA["along_spacing_m"])
)
route_length_m = float(chainage[-1])
terrain = _Terrain(project_root, filter_key, method, smooth)
valley_xy = (
_valley_points(project_root, filter_key, method, smooth)
if option in ("stability", "valley")
else np.empty((0, 2))
)
canopy = _canopy_grid(project_root) if option == "clearing" else None
result: dict[str, Any] = {
"option": option,
"spoil": [],
"borrow": [],
# 후보가 모자라 어느 부지에도 담지 못한 토량. 0이 아니면 화면이 그대로 알려야 한다 —
# 조용히 빠지면 사용자는 계획이 완성된 줄 알고 다음 단계로 넘어간다.
"unassigned_spoil_m3": 0.0,
"unassigned_borrow_m3": 0.0,
}
plans = (
("spoil", surplus_m3, _zone_center(surplus_zones or [])),
("borrow", shortage_m3, _zone_center(shortage_zones or [])),
)
for kind, demand, focus in plans:
if demand <= 0:
continue
candidates = _build_candidates(
option=option,
kind=kind,
terrain=terrain,
samples=samples,
sample_z=sample_z,
chainages=chainages,
tangent=tangent,
valley_xy=valley_xy,
canopy=canopy,
focus_m=focus,
route_length_m=route_length_m,
)
picked = _pick(candidates, demand)
result[kind] = picked
assigned = sum(float(site["assigned_m3"]) for site in picked)
result[f"unassigned_{kind}_m3"] = round(max(demand - assigned, 0.0), 1)
result["route_length_m"] = round(route_length_m, 2)
return result