refactor(476-1 ②): 측점 · 1차 계획선 셈과 파일 쓰기 나눔(동작 같음)
- Engine_Sections `build_sections`(측점 틀 · 지반 · 438 1차 계획선 · 포장 제안 · 파일 안 씀) · run_section_generation = 읽기 → 셈 → 쓰기 - 횡단 이름 규칙 → 새 `B05_Profile_Engine_Sections_Names`(700줄 · 다시 내보냄) - resolve_extra_stations 사본 칸(structures · plus) · 따라감 셈 몫 `followed_structures` · `followed_plus_rows` · `followed_pipes` · `section_seed` · `seat_base_pvi` 뗌 - 체인 · 추천 초기 종단이 같은 조각을 부를 자리 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -23,6 +23,17 @@ from B05_Profile.B05_Profile_Engine_Sections_Core import (
|
||||
SectionGenerationOptions,
|
||||
generate_sections,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Engine_Sections_Names import ( # noqa: F401 — 부르는 곳이 여기서 씀
|
||||
CROSS_KEY_PATTERN,
|
||||
cross_filename,
|
||||
cross_files_in_order,
|
||||
cross_key,
|
||||
cross_station,
|
||||
find_cross_file,
|
||||
parse_cross_key,
|
||||
prune_stale_cross_files,
|
||||
valid_cross_names,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Structures_Repository import load_structures
|
||||
from B05_Profile.B05_Profile_Structures_Schema import (
|
||||
structure_type_map,
|
||||
@@ -118,82 +129,6 @@ def _read_design_curve_table(project_root: Path) -> dict[str, Any]:
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ────────────
|
||||
#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80`
|
||||
CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)"
|
||||
|
||||
|
||||
def cross_station(chainage_m: float) -> int | float:
|
||||
"""측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리."""
|
||||
value = round(float(chainage_m), 2)
|
||||
whole = round(value)
|
||||
return int(whole) if abs(value - whole) < 0.005 else value
|
||||
|
||||
|
||||
def cross_key(chainage_m: float) -> str:
|
||||
"""이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점)."""
|
||||
station = cross_station(chainage_m)
|
||||
return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}"
|
||||
|
||||
|
||||
def parse_cross_key(text: str) -> int | float:
|
||||
"""`cross_key` 되읽기."""
|
||||
return cross_station(float(text))
|
||||
|
||||
|
||||
def cross_filename(chainage_m: float) -> str:
|
||||
"""측점 chainage에 대응하는 횡단면 파일명(단일 규칙)."""
|
||||
return f"cross_{cross_key(chainage_m)}m.json"
|
||||
|
||||
|
||||
def find_cross_file(cross_dir: Path, chainage_m: float) -> Path:
|
||||
"""읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트)."""
|
||||
path = cross_dir / cross_filename(chainage_m)
|
||||
legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json"
|
||||
return path if path.exists() or not legacy.exists() else legacy
|
||||
|
||||
|
||||
def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]:
|
||||
"""stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀).
|
||||
|
||||
측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다).
|
||||
"""
|
||||
# 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림
|
||||
return {
|
||||
find_cross_file(cross_dir, station["chainage_m"]).name
|
||||
for station in stations
|
||||
if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float))
|
||||
}
|
||||
|
||||
|
||||
def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]:
|
||||
"""잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞)."""
|
||||
valid = valid_cross_names(cross_dir, stations)
|
||||
|
||||
def chainage(path: Path) -> tuple[float, str]:
|
||||
match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem)
|
||||
return (float(match.group(1)) if match else float("inf"), path.name)
|
||||
|
||||
files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid]
|
||||
return sorted(files, key=chainage)
|
||||
|
||||
|
||||
def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
|
||||
"""stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다.
|
||||
|
||||
쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`.
|
||||
측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
|
||||
빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
|
||||
"""
|
||||
valid = valid_cross_names(cross_dir, stations)
|
||||
if not valid:
|
||||
return valid
|
||||
for path in cross_dir.glob("cross_*.json"):
|
||||
if path.name not in valid:
|
||||
path.unlink(missing_ok=True)
|
||||
return valid
|
||||
|
||||
|
||||
def _cross_summary(cross_section: dict[str, Any]) -> dict[str, Any]:
|
||||
"""횡단면 상세에서 DB data 컬럼에 저장할 요약을 만든다."""
|
||||
samples = cross_section.get("samples", [])
|
||||
@@ -297,7 +232,12 @@ def relabel_pipe_stations(
|
||||
|
||||
|
||||
def resolve_extra_stations(
|
||||
project_root: Path, pipes: list[PipePoint], xy: Sequence[Sequence[float]] | None = None
|
||||
project_root: Path,
|
||||
pipes: list[PipePoint],
|
||||
xy: Sequence[Sequence[float]] | None = None,
|
||||
*,
|
||||
structures: list[Any] | None = None,
|
||||
plus: list[float] | None = None,
|
||||
) -> tuple[tuple[float, str], ...]:
|
||||
"""구조물(비정규) 측점의 **단일 공급자** — 관 정본 + 구조물 정본 A군에서 파생한다.
|
||||
|
||||
@@ -314,12 +254,14 @@ def resolve_extra_stations(
|
||||
거울 시험 = `resources/tester/test_322_3_plan_mirror.py` · `test_272_1_station_list.py`.
|
||||
추가 측점(368-2) = `plus_stations.json` · 이름 "".
|
||||
`xy` = 지금 노선 — 노선 밖(시점 앞 · 종점 뒤 · `route_outside`) 관 · 구조물은 뺌(459).
|
||||
`structures` · `plus` = 파일 대신 따라간 사본(476 추천 초기 종단 · 파일 안 씀).
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Plus_Stations import plus_chainages
|
||||
from common_util.common_util_route_follow import route_outside
|
||||
|
||||
try:
|
||||
structures = load_structures(str(project_root))[1]
|
||||
if structures is None:
|
||||
structures = load_structures(str(project_root))[1]
|
||||
except Exception:
|
||||
logger.exception("B05: 구조물 정본을 읽지 못했습니다 (관 유래 측점만 씁니다)")
|
||||
structures = []
|
||||
@@ -332,7 +274,9 @@ def resolve_extra_stations(
|
||||
]
|
||||
return tuple(
|
||||
(float(station["chainage_m"]), str(station["title"]))
|
||||
for station in plan_stations(pipes, structures, plus_chainages(project_root))
|
||||
for station in plan_stations(
|
||||
pipes, structures, plus_chainages(project_root) if plus is None else plus
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@@ -450,6 +394,55 @@ def _annotate_pavement_suggestions(
|
||||
station["pavement_grade_limit_pct"] = round(unpaved_limit, 2)
|
||||
|
||||
|
||||
def build_sections(
|
||||
polyline: list[list[float]],
|
||||
sampler: Any,
|
||||
pipes: list[PipePoint],
|
||||
extras: tuple[tuple[float, str], ...],
|
||||
*,
|
||||
options: SectionGenerationOptions | None,
|
||||
grade_options: GradeDesignOptions | None,
|
||||
source_snapshot: dict[str, Any],
|
||||
crs: str | None,
|
||||
design_curves: list[dict[str, Any]],
|
||||
station_snap: bool,
|
||||
station_snap_step: float | None,
|
||||
) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
||||
"""측점 틀 · 지반 · 1차 계획선(438) · 포장 제안 — 파일 안 씀 · (결과, 계획선 요약).
|
||||
|
||||
종횡단 생성(`run_section_generation`)과 추천 후보 초기 종단(476 · `..._RouteInitial_Initial`)
|
||||
이 같은 셈 한 벌."""
|
||||
result = generate_sections(
|
||||
polyline,
|
||||
sampler,
|
||||
replace(options or SectionGenerationOptions(), extra_stations=extras),
|
||||
source_snapshot=source_snapshot,
|
||||
crs=crs,
|
||||
design_curves=design_curves,
|
||||
station_snap=station_snap,
|
||||
station_snap_step=station_snap_step,
|
||||
)
|
||||
grade_summary = _append_design_profiles(
|
||||
result["longitudinal"],
|
||||
grade_options,
|
||||
(result.get("options") or {}).get("station_interval_m"),
|
||||
# 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록).
|
||||
pipe_chainages=[pipe.chainage_m for pipe in pipes],
|
||||
# 시설 제원 최소 여유(관경+토피 등)로 계획선 들어올림 — 사용자 스위치(기본 해제 · 09-01).
|
||||
pipe_clearances=(
|
||||
dict(pipe_anchor_clearances(pipes))
|
||||
if grade_options and grade_options.enforce_pipe_clearance
|
||||
else None
|
||||
),
|
||||
)
|
||||
# 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적.
|
||||
try:
|
||||
_annotate_pavement_suggestions(result["longitudinal"], grade_options)
|
||||
except Exception:
|
||||
logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)")
|
||||
return result, grade_summary
|
||||
|
||||
|
||||
def run_section_generation(
|
||||
project_root: Path,
|
||||
route_data_path: str,
|
||||
@@ -487,10 +480,13 @@ def run_section_generation(
|
||||
extras = ()
|
||||
logger.info("B05 종횡단: 구조물 측점 %d건 파생(관 %d건)", len(extras), len(pipes))
|
||||
|
||||
result = generate_sections(
|
||||
result, grade_summary = build_sections(
|
||||
polyline,
|
||||
sampler,
|
||||
replace(options or SectionGenerationOptions(), extra_stations=extras),
|
||||
pipes,
|
||||
extras,
|
||||
options=options,
|
||||
grade_options=grade_options,
|
||||
source_snapshot={"filter": filter_key, "method": method, "smooth": smooth},
|
||||
crs=crs,
|
||||
# 곡선부 확폭의 정본 — 설계 곡선표(시·종점·반지름)를 그대로 쓴다. 없으면 빈 목록이라
|
||||
@@ -505,29 +501,6 @@ def run_section_generation(
|
||||
cross_dir = stage_root / "cross_sections"
|
||||
long_dir.mkdir(parents=True, exist_ok=True)
|
||||
cross_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 종단면 저장 (계획선은 저장 직전에 종단 데이터에 붙인다)
|
||||
grade_summary = _append_design_profiles(
|
||||
result["longitudinal"],
|
||||
grade_options,
|
||||
(result.get("options") or {}).get("station_interval_m"),
|
||||
# 1차 계획선의 변화점 = 배수유역도가 확정한 배관 배치 측점(측점 생성과 같은 목록).
|
||||
pipe_chainages=[pipe.chainage_m for pipe in pipes],
|
||||
# 시설 제원이 요구하는 최소 여유(관경+토피 등) — 계획선이 그만큼 들려야 관·구체가
|
||||
# 들어갈 자리가 생긴다(2026-08-23 사용자 지시). 다만 이 들어올림이 배수 자리마다
|
||||
# 계획고를 잡아당겨 제어가 어려워, 적용 여부를 사용자 스위치로 뺐다(2026-09-01
|
||||
# 사용자 지시 — 기본 해제). 산식은 그대로이고 켜고 끄는 것만 바뀐다.
|
||||
pipe_clearances=(
|
||||
dict(pipe_anchor_clearances(pipes))
|
||||
if grade_options and grade_options.enforce_pipe_clearance
|
||||
else None
|
||||
),
|
||||
)
|
||||
# 계획선 경사 기반 포장 제안(법정 상한 초과 측점) — 비치명적.
|
||||
try:
|
||||
_annotate_pavement_suggestions(result["longitudinal"], grade_options)
|
||||
except Exception:
|
||||
logger.exception("B05 포장 제안 판정 실패 (종횡단은 유지)")
|
||||
long_file = long_dir / "longitudinal.json"
|
||||
atomic_write_json(long_file, result["longitudinal"])
|
||||
long_summary = {
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""횡단 이름 규칙 한 벌 — 파일 이름 · B07 도면 id · 찾기 · 잔재 거름.
|
||||
|
||||
`B05_Profile_Engine_Sections` 에서 나눔(700줄) · 그 모듈이 다시 내보냄."""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
# ── 횡단 이름 규칙(한 벌) — 파일 이름 · B07 도면 id · 찾기가 모두 이것을 씀 ────────────
|
||||
#: 이름 속 측점 글 — 정수 m `00120` · 소수 측점 `00119.80`
|
||||
CROSS_KEY_PATTERN = r"(\d+(?:\.\d{2})?)"
|
||||
|
||||
|
||||
def cross_station(chainage_m: float) -> int | float:
|
||||
"""측점 → 이름 열쇠 값 — 정수 m(0.005m 안)면 int(옛 이름 · 키 그대로) · 아니면 소수 2자리."""
|
||||
value = round(float(chainage_m), 2)
|
||||
whole = round(value)
|
||||
return int(whole) if abs(value - whole) < 0.005 else value
|
||||
|
||||
|
||||
def cross_key(chainage_m: float) -> str:
|
||||
"""이름 속 측점 글 — `00120`(정수 · 옛 이름 그대로) · `00119.80`(소수 측점)."""
|
||||
station = cross_station(chainage_m)
|
||||
return f"{station:05d}" if isinstance(station, int) else f"{station:08.2f}"
|
||||
|
||||
|
||||
def parse_cross_key(text: str) -> int | float:
|
||||
"""`cross_key` 되읽기."""
|
||||
return cross_station(float(text))
|
||||
|
||||
|
||||
def cross_filename(chainage_m: float) -> str:
|
||||
"""측점 chainage에 대응하는 횡단면 파일명(단일 규칙)."""
|
||||
return f"cross_{cross_key(chainage_m)}m.json"
|
||||
|
||||
|
||||
def find_cross_file(cross_dir: Path, chainage_m: float) -> Path:
|
||||
"""읽기용 — 지금 이름이 없으면 옛 정수 m 이름(소수 측점이 정수로 붙던 옛 프로젝트)."""
|
||||
path = cross_dir / cross_filename(chainage_m)
|
||||
legacy = cross_dir / f"cross_{int(round(float(chainage_m))):05d}m.json"
|
||||
return path if path.exists() or not legacy.exists() else legacy
|
||||
|
||||
|
||||
def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]:
|
||||
"""stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀).
|
||||
|
||||
측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다).
|
||||
"""
|
||||
# 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림
|
||||
return {
|
||||
find_cross_file(cross_dir, station["chainage_m"]).name
|
||||
for station in stations
|
||||
if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float))
|
||||
}
|
||||
|
||||
|
||||
def cross_files_in_order(cross_dir: Path, stations: list[Any]) -> list[Path]:
|
||||
"""잔재를 걸러 읽을 cross_*.json — **측점 차례**(이름 차례면 `01080.16` 이 `01080` 앞)."""
|
||||
valid = valid_cross_names(cross_dir, stations)
|
||||
|
||||
def chainage(path: Path) -> tuple[float, str]:
|
||||
match = re.fullmatch(rf"cross_{CROSS_KEY_PATTERN}m", path.stem)
|
||||
return (float(match.group(1)) if match else float("inf"), path.name)
|
||||
|
||||
files = [path for path in cross_dir.glob("cross_*.json") if not valid or path.name in valid]
|
||||
return sorted(files, key=chainage)
|
||||
|
||||
|
||||
def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
|
||||
"""stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다.
|
||||
|
||||
쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`.
|
||||
측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
|
||||
빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
|
||||
"""
|
||||
valid = valid_cross_names(cross_dir, stations)
|
||||
if not valid:
|
||||
return valid
|
||||
for path in cross_dir.glob("cross_*.json"):
|
||||
if path.name not in valid:
|
||||
path.unlink(missing_ok=True)
|
||||
return valid
|
||||
@@ -75,6 +75,14 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo
|
||||
rows = load_plus_stations(project_root)
|
||||
if not rows:
|
||||
return 0
|
||||
followed = followed_plus_rows(rows, new_xy)
|
||||
if followed["moved"]:
|
||||
atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]})
|
||||
return int(followed["moved"])
|
||||
|
||||
|
||||
def followed_plus_rows(rows: list[dict[str, Any]], new_xy: Sequence[Sequence[float]]) -> dict:
|
||||
"""추가 측점을 새 노선에 투영(저장 안 함) → `{rows, moved}` — 체인 · 추천 초기 종단(476)."""
|
||||
results = run_calc(
|
||||
["facility.plus_follow"],
|
||||
{
|
||||
@@ -89,7 +97,4 @@ def follow_plus_stations(project_root: str | Path, new_xy: Sequence[Sequence[flo
|
||||
)
|
||||
if results is None:
|
||||
raise RuntimeError("추가 측점 따라감 실패 — 계산 Node 입구(facility.plus_follow)")
|
||||
followed = results["facility.plus_follow"]
|
||||
if followed["moved"]:
|
||||
atomic_write_json(str(plus_stations_path(project_root)), {"stations": followed["rows"]})
|
||||
return int(followed["moved"])
|
||||
return results["facility.plus_follow"]
|
||||
|
||||
@@ -22,7 +22,7 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan
|
||||
(`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`).
|
||||
- 계획고 = 바뀐 자리만 새 지반 첫 계획선 · 같은 자리(옆 1 m 안) 사용자 계획고 지킴
|
||||
(`B05_Profile_Engine_Grade_Carry` · 옛 · 새 **계획 폴리라인** — 화면 planline 과 같은 선) ·
|
||||
추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) 또는 1차 계획선(438).
|
||||
추천 [적용](`as_initial`) = 옮기지 않음 — 균형 종단(`apply_profile_target`) · 1차 계획선(438).
|
||||
- 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]).
|
||||
"""
|
||||
|
||||
@@ -165,6 +165,31 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
|
||||
return []
|
||||
|
||||
|
||||
def followed_pipes(
|
||||
old_pipes: list, old_xy: list[list[float]], points: list[dict[str, float]], streams: list
|
||||
) -> tuple[list, int, list]:
|
||||
"""관 따라감 + 새 세류선 교차점 기본 관(저장 안 함) → (짝, 교차 수, 새 노선 정점) — 체인 ·
|
||||
추천 초기 종단(476) 한 벌."""
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
|
||||
from common_util.common_util_drainage_pipes import fill_pipe_coordinates
|
||||
|
||||
vertices = vertices_of([(p["x"], p["y"]) for p in points])
|
||||
fill_pipe_coordinates(old_pipes, vertices_of(old_xy))
|
||||
pairs = follow_pipes(old_pipes, vertices)
|
||||
total = float(vertices[-1].chainage_m)
|
||||
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
|
||||
return pairs, crossing, vertices
|
||||
|
||||
|
||||
def section_seed(request: Any, old_section_options: dict[str, Any] | None) -> dict[str, Any]:
|
||||
"""체인 종횡단 생성 옵션(건너뛴 첫 생성이 남겼을 값) — 체인 · 추천 초기 종단(476) 한 벌."""
|
||||
from B05_Profile.B05_Profile_Router import _section_options
|
||||
|
||||
seed = asdict(_section_options(request, old_section_options))
|
||||
seed.pop("extra_stations", None)
|
||||
return seed
|
||||
|
||||
|
||||
async def _place_pipes(
|
||||
project_id: UUID,
|
||||
stored_path: str,
|
||||
@@ -178,22 +203,13 @@ async def _place_pipes(
|
||||
노선은 범위를 안 잘라 누가거리가 갈림 · 세류선만 거기서. 좌표 없는 옛 관은 옛 노선
|
||||
자리로 먼저 채움(459).
|
||||
"""
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
|
||||
from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare
|
||||
from common_util.common_util_drainage_pipes import (
|
||||
fill_pipe_coordinates,
|
||||
route_signature,
|
||||
save_pipe_points,
|
||||
)
|
||||
from common_util.common_util_drainage_pipes import route_signature, save_pipe_points
|
||||
from common_util.common_util_route_follow import route_outside
|
||||
|
||||
prepared = await _prepare(project_id)
|
||||
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음
|
||||
vertices = vertices_of([(p["x"], p["y"]) for p in points])
|
||||
fill_pipe_coordinates(old_pipes, vertices_of(old_xy))
|
||||
pairs = follow_pipes(old_pipes, vertices)
|
||||
total = float(vertices[-1].chainage_m)
|
||||
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
|
||||
pairs, crossing, vertices = followed_pipes(old_pipes, old_xy, points, streams)
|
||||
await asyncio.to_thread(
|
||||
save_pipe_points,
|
||||
stored_path,
|
||||
@@ -259,7 +275,7 @@ async def _carry_profile(
|
||||
|
||||
|
||||
def carries_cross_inputs(initial: bool) -> bool:
|
||||
"""횡단 손 입력을 새 측점(같은 자리)에 넘기나 — 추천 [적용]도 넘김(475-3 · 사용자 답 대기 476)."""
|
||||
"""횡단 손 입력을 새 측점(같은 자리)에 넘기나 — [적용]도 넘김(475-3 · 사용자 답 대기 476)."""
|
||||
return True
|
||||
|
||||
|
||||
@@ -376,11 +392,7 @@ async def run_route_save_chain(
|
||||
from B03_FileInput.B03_FileInput_Service_Chain import _log_steps, redesign_solve_request
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _STAGE_SUBDIR
|
||||
from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations, plus_stations_path
|
||||
from B05_Profile.B05_Profile_Router import (
|
||||
_section_options,
|
||||
confirm_latest_route,
|
||||
solve_route_record,
|
||||
)
|
||||
from B05_Profile.B05_Profile_Router import confirm_latest_route, solve_route_record
|
||||
from B05_Profile.B05_Profile_Structures_Repository import load_structures, structures_file_path
|
||||
from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
|
||||
from B06_Section.B06_Section_Repository import (
|
||||
@@ -456,8 +468,7 @@ async def run_route_save_chain(
|
||||
pairs, pipes = await _place_pipes(project_id, stored_path, old_pipes, old_xy, points)
|
||||
marks.append(("관 따라감 · 교차점", time.perf_counter()))
|
||||
|
||||
seed = asdict(_section_options(request, old_section_options))
|
||||
seed.pop("extra_stations", None)
|
||||
seed = section_seed(request, old_section_options)
|
||||
made = await regenerate_route_sections(
|
||||
project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed
|
||||
)
|
||||
|
||||
@@ -355,15 +355,22 @@ async def refresh_expected_route(project_id: UUID) -> None:
|
||||
|
||||
def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
|
||||
"""종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다."""
|
||||
import json
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
|
||||
longitudinal = json.loads(path.read_text(encoding="utf-8"))
|
||||
count = seat_base_pvi(longitudinal, stations, plan)
|
||||
atomic_write_json(path, longitudinal)
|
||||
return count
|
||||
|
||||
|
||||
def seat_base_pvi(longitudinal: dict[str, Any], stations: list[float], plan: list[float]) -> int:
|
||||
"""1차 변화점 계획고를 균형 종단에 얹음(제자리 · 파일 안 씀) — [적용] 체인 · 추천 초기 종단(476)
|
||||
한 벌."""
|
||||
import numpy as np
|
||||
|
||||
from B05_Profile.B05_Profile_Engine_Grade_Alignment import AlignmentPolicy
|
||||
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import held_pass_through
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
|
||||
longitudinal = json.loads(path.read_text(encoding="utf-8"))
|
||||
alignment = longitudinal["profile_alignment"]
|
||||
base = alignment["base_pvi"]
|
||||
base_s = np.array([float(item["chainage_m"]) for item in base])
|
||||
@@ -374,7 +381,6 @@ def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
|
||||
)
|
||||
for item, value in zip(base, base_z, strict=True):
|
||||
item["elevation_m"] = round(float(value), 6) # 직선 판정(426) 1e-5 m 안
|
||||
atomic_write_json(path, longitudinal)
|
||||
return len(base)
|
||||
|
||||
|
||||
|
||||
@@ -121,6 +121,26 @@ async def _route_xy(project_id: UUID, root: str) -> list[list[float]] | None:
|
||||
return None
|
||||
|
||||
|
||||
def followed_structures(
|
||||
project_root: Path,
|
||||
items: list[Any],
|
||||
old_route_data_path: str | None,
|
||||
new_xy: list[dict[str, float]],
|
||||
) -> list[Any]:
|
||||
"""구조물을 새 노선 땅 위 자리로 옮김(제자리 · 저장 안 함) — 노선 저장 체인 · 추천 초기
|
||||
종단(476) 한 벌. 좌표 없는 옛 자료는 옛 노선으로 먼저 채움."""
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
|
||||
|
||||
if old_route_data_path:
|
||||
try:
|
||||
fill_structure_coordinates(
|
||||
items, _load_route_polyline(project_root, old_route_data_path), only_missing=True
|
||||
)
|
||||
except (OSError, ValueError):
|
||||
logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로")
|
||||
return follow_structures(items, [(point["x"], point["y"]) for point in new_xy])
|
||||
|
||||
|
||||
def follow_structures_to_route(
|
||||
project_root: Path, old_route_data_path: str | None, new_xy: list[dict[str, float]]
|
||||
) -> dict[str, int]:
|
||||
@@ -129,21 +149,12 @@ def follow_structures_to_route(
|
||||
좌표가 없는 옛 자료는 옛 노선(아직 안 바뀐 경로 파일)으로 먼저 채운다. 실패는 비치명 —
|
||||
구조물은 옛 누가거리 그대로 남는다(지금까지와 같음).
|
||||
"""
|
||||
from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
|
||||
|
||||
root = str(project_root)
|
||||
revision, items = load_structures(root)
|
||||
if not items:
|
||||
return {"moved": 0, "far": 0}
|
||||
if old_route_data_path:
|
||||
try:
|
||||
fill_structure_coordinates(
|
||||
items, _load_route_polyline(project_root, old_route_data_path), only_missing=True
|
||||
)
|
||||
except (OSError, ValueError):
|
||||
logger.warning("B05 구조물 따라감: 옛 노선을 못 읽어 좌표 없는 구조물은 그대로")
|
||||
before = [(item.chainage_m, item.start_m, item.end_m) for item in items]
|
||||
result = follow_structures(items, [(point["x"], point["y"]) for point in new_xy])
|
||||
result = followed_structures(project_root, items, old_route_data_path, new_xy)
|
||||
moved = sum(
|
||||
before[index] != (item.chainage_m, item.start_m, item.end_m)
|
||||
for index, item in enumerate(items)
|
||||
|
||||
Reference in New Issue
Block a user