eomsangdonandClaude Opus 5.5
7e1b5bd27e
- 설계 계산 ford_paved(파이썬) · fordPaved(TS 짝) — 물넘이 범위 안 측점은 측구가 안 섬(사용자 측구 선택보다 먼저) - 서버 정본 — ford_pavement_ranges 로 기본 설계 · [저장] 재계산 · 포장 범위 맞춤이 판정 · 저장분 재계산은 되실은 값을 이음 - 화면 — 계획선 편집 재계산이 얹힌 물넘이 제원으로 같은 판정 - 노면 기울기 — 유입 = 설계선 양 끝 표고가 높은 쪽(같으면 단면유형) · 경사각도만큼 낮은 쪽(유출)으로 · 도수로 띠도 같은 쪽 - 시험 test_286_1 · 횡단 설계 거울에 물넘이 갈래 · 도수로 띠 시험 기대값 새 규칙으로 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TteK8mfYZetk5P3bEZkyVZ
460 lines
21 KiB
Python
460 lines
21 KiB
Python
"""B05 경로 확정 시 종단 정본 파일에 비정규 측점·상단측 변경을 병합하는 헬퍼.
|
|
|
|
라우터(`B05_Profile_Router.py`)의 confirm 엔드포인트가 호출한다. 파일 기반 병합이라
|
|
재확정해도 중복이 생기지 않으며, 모든 호출은 비치명적(실패해도 경로 확정은 진행)이다.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Engine_Grade_Alignment import chainage_key
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
|
|
from B05_Profile.B05_Profile_Engine_Sections import (
|
|
_load_pipe_points,
|
|
_load_route_polyline,
|
|
find_cross_file,
|
|
generate_irregular_sections,
|
|
resolve_extra_stations,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
|
|
from B05_Profile.B05_Profile_Repository import (
|
|
get_surface_crs_epsg,
|
|
sync_cross_section_rows,
|
|
update_longitudinal_grade_summary,
|
|
)
|
|
from B05_Profile.B05_Profile_Schema import RouteConfirmRequest
|
|
from B06_Section.B06_Section_Repository import (
|
|
get_latest_section_options,
|
|
get_longitudinal_section,
|
|
)
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
|
|
|
|
def _section_options_from_stored(stored: dict[str, Any] | None) -> SectionGenerationOptions:
|
|
"""저장된 종횡단 옵션(단일 소스)만으로 옵션을 재구성한다(확정 시 비정규 측점 샘플링용)."""
|
|
defaults = SectionGenerationOptions()
|
|
stored = stored or {}
|
|
return SectionGenerationOptions(
|
|
station_interval_m=stored.get("station_interval_m") or defaults.station_interval_m,
|
|
cross_half_width_m=stored.get("cross_half_width_m") or defaults.cross_half_width_m,
|
|
cross_sample_interval_m=stored.get("cross_sample_interval_m")
|
|
or defaults.cross_sample_interval_m,
|
|
long_sample_interval_m=stored.get("long_sample_interval_m")
|
|
or defaults.long_sample_interval_m,
|
|
include_endpoint=defaults.include_endpoint,
|
|
)
|
|
|
|
|
|
def _merge_uphill_overrides_into_longitudinal(
|
|
project_root: Path, longitudinal_file_path: str, overrides: list[dict[str, Any]]
|
|
) -> None:
|
|
"""종단 정본 파일 stations의 uphill_side를 사용자 변경값으로 덮어쓴다.
|
|
|
|
solve가 자동 판정한 상단측(=측구 방향)을 3D 램프 클릭으로 바꾼 경우, 확정 시점에
|
|
정본에 반영해 B06이 값만 읽으면 되게 한다. 사용자 지정임을 소스 필드로 남긴다.
|
|
"""
|
|
if not overrides:
|
|
return
|
|
root = project_root.resolve()
|
|
path = (root / longitudinal_file_path).resolve()
|
|
if root not in path.parents or not path.is_file():
|
|
return
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
stations = data.get("stations")
|
|
if not isinstance(stations, list):
|
|
return
|
|
by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
|
|
changed = False
|
|
for station in stations:
|
|
side = by_chainage.get(round(float(station.get("chainage_m", -1.0)), 3))
|
|
if side is None:
|
|
continue
|
|
station["uphill_side"] = side
|
|
station["uphill_side_source"] = "user"
|
|
changed = True
|
|
if changed:
|
|
data["stations"] = stations
|
|
atomic_write_json(path, data)
|
|
|
|
|
|
async def sync_uphill_overrides_into_designs(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
route_id: int,
|
|
project_root: Path,
|
|
longitudinal_file_path: str,
|
|
overrides: list[dict[str, Any]],
|
|
) -> None:
|
|
"""상단측 변경 측점의 **저장된 횡단 설계**를 새 방향으로 재계산해 저장한다.
|
|
|
|
종단 정본(stations.uphill_side)만 바꾸면 저장 설계(design.section_mode/ditch_side)가
|
|
옛 방향으로 남아 B06이 그것을 계속 표시하고, 종횡단 확정의 역반영
|
|
(ditch_side→uphill_side)이 사용자 지정을 다시 옛 방향으로 되돌린다
|
|
(2026-08-06 13측점 보고: uphill=right vs design=left_cut 순환 덮어쓰기).
|
|
편절은 절토측을, 양절은 측구측을 새 방향으로 맞추고 기하를 즉시 재계산한다.
|
|
양성(both_fill)은 측구가 없으므로 건드리지 않는다.
|
|
"""
|
|
# 지연 import — B06 라우터 모듈 로드는 이 함수가 실제 불릴 때만 필요하다.
|
|
from B06_Section.B06_Section_Engine_Design import (
|
|
compute_cross_design,
|
|
curve_widening_args,
|
|
stored_shoulder_extend,
|
|
)
|
|
from B06_Section.B06_Section_Repository import get_cross_section_designs
|
|
from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
|
|
from B06_Section.B06_Section_Router_Design import (
|
|
ford_drop_at,
|
|
ford_surface_drops,
|
|
pavement_suggestions,
|
|
read_cross_design_inputs,
|
|
resolve_longitudinal_path,
|
|
)
|
|
|
|
if not overrides:
|
|
return
|
|
by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
|
|
designs = await get_cross_section_designs(connection, route_id)
|
|
# 표준단면 수치는 종단 정본 options에 병합 저장돼 있다(종횡단 확정 시) — 없으면 기본값.
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route_id)
|
|
stored_standard = None
|
|
if longitudinal and isinstance(longitudinal.get("data"), dict):
|
|
options = longitudinal["data"].get("options")
|
|
if isinstance(options, dict):
|
|
stored_standard = options.get("standard_cross_section")
|
|
# 다시 계산할 측점만 먼저 고른다 — 파일·계산은 아래에서 **스레드 한 번**에 몰아 한다.
|
|
jobs: list[tuple[float, dict[str, Any], str, str]] = []
|
|
for record in designs:
|
|
chainage = round(float(record["chainage_m"]), 3)
|
|
side = by_chainage.get(chainage)
|
|
design = record.get("design")
|
|
if side is None or not isinstance(design, dict):
|
|
continue
|
|
mode = design.get("section_mode")
|
|
if mode == "both_fill":
|
|
continue
|
|
next_mode = f"{side}_cut" if mode in ("left_cut", "right_cut") else mode
|
|
if next_mode == mode and design.get("ditch_side") == side:
|
|
continue
|
|
jobs.append((chainage, design, str(next_mode), side))
|
|
if not jobs:
|
|
return
|
|
|
|
def _recompute() -> list[tuple[float, dict[str, Any]]]:
|
|
"""측점마다 종단 정본을 다시 열던 것을 한 번으로 줄인다(측점당 13.8ms 였다)."""
|
|
root = Path(project_root)
|
|
longitudinal_path = resolve_longitudinal_path(root, longitudinal_file_path)
|
|
longitudinal_json = json.loads(longitudinal_path.read_text(encoding="utf-8"))
|
|
preloaded = (longitudinal_path, longitudinal_json, pavement_suggestions(longitudinal_json))
|
|
ford_drops = ford_surface_drops(root)
|
|
out: list[tuple[float, dict[str, Any]]] = []
|
|
for chainage, design, next_mode, side in jobs:
|
|
samples, design_elevation, pavement_suggested, cross_record = read_cross_design_inputs(
|
|
root, longitudinal_file_path, float(chainage), preloaded
|
|
)
|
|
next_design = compute_cross_design(
|
|
samples,
|
|
design_elevation,
|
|
ground_type=str(design.get("ground_type", "soil")),
|
|
section_mode=next_mode,
|
|
ditch_side=side,
|
|
ditch_type=str(design.get("ditch_type", "standard")),
|
|
paved=bool(design.get("paved", False)),
|
|
standard=stored_standard,
|
|
rock_boundary_offset_m=design.get("rock_boundary_offset_m"),
|
|
two_stage_slope=bool(design.get("two_stage_slope", True)),
|
|
ditch_enabled=design.get("ditch_enabled"),
|
|
ditch_choice=design.get("ditch_choice"),
|
|
surface_drop_m=ford_drop_at(float(chainage), ford_drops),
|
|
**curve_widening_args(cross_record),
|
|
**stored_shoulder_extend(design), # 노견 확장(139-6)
|
|
ford_paved=bool(design.get("ford_paved")), # 물넘이 자리 측구 없음(286-2)
|
|
)
|
|
next_design["status"] = design.get("status", "provisional")
|
|
next_design["pavement_suggested"] = design.get("pavement_suggested", pavement_suggested)
|
|
# 개별 표시 반폭 등 계산과 무관한 표시 설정은 그대로 이월한다.
|
|
if design.get("display_half_width_m") is not None:
|
|
next_design["display_half_width_m"] = design["display_half_width_m"]
|
|
out.append((float(chainage), next_design))
|
|
return out
|
|
|
|
# 쓰기도 한 문장으로 — 행마다 내면 원격 DB 왕복이 측점 수만큼 난다(건당 24.5ms).
|
|
await merge_cross_section_designs(
|
|
connection,
|
|
route_id=route_id,
|
|
entries=await asyncio.to_thread(_recompute),
|
|
replace=True,
|
|
project_id=project_id,
|
|
)
|
|
|
|
|
|
#: 반올림만 다른 같은 측점(m) — 서버가 관 측점을 0.01m 로 반올림.
|
|
#: 화면 `applyIrregularStations` 0.006 과 짝.
|
|
_SAME_STATION_M = 0.006
|
|
|
|
|
|
def _follow_moved_stations(
|
|
data: dict[str, Any], before: list[float], previous: list[float] | None = None
|
|
) -> dict[str, Any] | None:
|
|
"""사라진 비정규 측점의 계획선 꼭짓점(base_pvi) · 편집을 새 측점으로 옮기거나 지운다.
|
|
|
|
남기면 측점선 없는 변화점이 계획선을 꺾는다(192-1 — 관 769.28 → 769.38 뒤 두 자리에
|
|
편집이 남아 0.1m 사이 −465% 계단). 반올림 차이(0.006m 안) · 옛 소수 측점의 정수 m 맞춤
|
|
자리(272)만 옮기고 나머지는 지운다.
|
|
측점 목록이 계획선 측점 줄과 달라져도 다시 센다(272). 다시 셌으면 새 등급 요약(DB 몫)을,
|
|
아니면 `None`.
|
|
"""
|
|
stored = data.get("profile_alignment")
|
|
if not isinstance(stored, dict) or not stored.get("base_pvi"):
|
|
return None
|
|
after = [float(station.get("chainage_m", 0.0)) for station in data["stations"]]
|
|
after_keys = {chainage_key(value) for value in after}
|
|
before_keys = {chainage_key(value) for value in before}
|
|
moves: dict[str, str | None] = {}
|
|
for old in before:
|
|
if chainage_key(old) in after_keys:
|
|
continue
|
|
near = next(
|
|
(
|
|
value
|
|
for value in after
|
|
if abs(value - old) < _SAME_STATION_M and chainage_key(value) not in before_keys
|
|
),
|
|
None,
|
|
)
|
|
# 옛 소수 측점 → 정수 m 맞춤(272 · 요구 167) — 맞춘 자리가 새 측점이면 꼭짓점 · 편집도
|
|
# 그리로(지우면 계획선이 그 자리에서 바뀜 · 화면 `followMovedNodes` 와 짝)
|
|
snapped = chainage_key(math.floor(old + 0.5))
|
|
if near is None and snapped in after_keys and snapped not in before_keys:
|
|
near = float(snapped)
|
|
moves[chainage_key(old)] = None if near is None else chainage_key(near)
|
|
|
|
def moved(value: float) -> tuple[bool, str | None]:
|
|
"""옛 측점 0.006m 안이면 (참, 옮길 키 · 지우면 None) — 옛 글자(263.206)도 같은 측점(213-1).
|
|
남은 측점 위 값은 그대로."""
|
|
if chainage_key(value) in after_keys:
|
|
return False, None
|
|
for old_key, new_key in moves.items():
|
|
if abs(float(old_key) - value) < _SAME_STATION_M:
|
|
return True, new_key
|
|
return False, None
|
|
|
|
edits = stored.get("edits") or {}
|
|
next_edits: dict[str, dict[str, Any]] = {}
|
|
changed = False
|
|
for name in ("station_offsets", "curve_radii"):
|
|
values = dict(edits.get(name) or {})
|
|
# 옛 측점 글자 그대로인 키가 먼저 — 반올림 글자보다 이김
|
|
for key in sorted(values, key=lambda item: item not in moves):
|
|
hit, new_key = moved(float(key))
|
|
if not hit:
|
|
continue
|
|
value = values.pop(key)
|
|
if new_key is not None:
|
|
values.setdefault(new_key, value)
|
|
changed = True
|
|
next_edits[name] = values
|
|
base: list[dict[str, Any]] = []
|
|
for node in stored["base_pvi"]:
|
|
hit, new_key = moved(float(node["chainage_m"]))
|
|
if not hit:
|
|
base.append(node)
|
|
continue
|
|
changed = True
|
|
if new_key is not None:
|
|
base.append({**node, "chainage_m": float(new_key)})
|
|
# 측점 목록만 바뀌어도 다시 셈 — 계획선 측점 줄(표 · ▲▼)이 옛 측점을 들고 남지 않게(272) ·
|
|
# 줄이 없는 판은 맞추기 전 측점(`previous`)과 견줌
|
|
rows = stored.get("stations")
|
|
listed = (
|
|
[round(float(row.get("chainage_m", 0.0)), 6) for row in rows]
|
|
if isinstance(rows, list)
|
|
else previous
|
|
)
|
|
if not changed and listed in (None, [round(value, 6) for value in after]):
|
|
return None
|
|
stored["base_pvi"] = base
|
|
alignment, profile = rebuild_alignment_profile(data, next_edits)
|
|
data["profile_alignment"] = alignment
|
|
profiles = data.get("design_profiles")
|
|
if isinstance(profiles, list) and profiles:
|
|
profiles[0] = profile
|
|
else:
|
|
data["design_profiles"] = [profile]
|
|
return {"id": profile["id"], **profile["summary"]}
|
|
|
|
|
|
def _merge_irregular_into_longitudinal(
|
|
project_root: Path, longitudinal_file_path: str, irregular_stations: list[dict[str, Any]]
|
|
) -> dict[str, Any] | None:
|
|
"""종단 정본 stations 의 구조물 측점을 **지금 목록 전체로 다시 맞춘다**(272 · 요구 167).
|
|
|
|
`irregular_stations` = 지금 관 · 구조물로 새로 뜬 구조물 측점 전부(격자와 같은 m 인 것은
|
|
`grid` 표시 · `generate_irregular_sections`). 옛 구조물 측점은 다 빼고(빈 목록이어도),
|
|
구조물이 떠난 격자 자리(`grid`)는 일반 측점으로 되돌리고, 새 목록의 격자 자리 측점은 그
|
|
격자 측점을 갈음한다(같은 m 측점 하나 · 구조물이 위). 표고 샘플은 손대지 않는다.
|
|
B06 상세는 종단 파일의 stations 에 맞는 cross 파일만 노출하고 남은 파일을 지운다
|
|
(`prune_stale_cross_files`). 계획선은 사라진 구조물 측점 자리만 고친다(`_follow_moved_stations`
|
|
— 바뀌면 새 등급 요약을 돌려줌). 파일 기반이라 다시 맞춰도 중복이 생기지 않는다.
|
|
TS 짝 = `common_util_structure_station.ts` `oneStationList`(화면 측점 한 벌) · 거울 시험 =
|
|
`resources/tester/test_272_1_station_list.py`.
|
|
"""
|
|
root = project_root.resolve()
|
|
path = (root / longitudinal_file_path).resolve()
|
|
if root not in path.parents or not path.is_file():
|
|
return None
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
stations = data.get("stations")
|
|
if not isinstance(stations, list):
|
|
return None
|
|
before = [
|
|
float(station.get("chainage_m", 0.0))
|
|
for station in stations
|
|
if station.get("kind") == "irregular"
|
|
]
|
|
fresh = {round(float(s.get("chainage_m", 0.0)), 6): s for s in irregular_stations}
|
|
kept: dict[float, dict[str, Any]] = {}
|
|
left: list[dict[str, Any]] = []
|
|
for station in stations:
|
|
if station.get("kind") == "irregular" and not station.get("grid"):
|
|
continue # 옛 구조물 측점 — 새 목록이 갈음
|
|
key = round(float(station.get("chainage_m", 0.0)), 6)
|
|
if key not in fresh and ("structure" in station or station.get("grid")):
|
|
# 구조물이 떠난 격자 · 시종점 자리 — 일반 측점으로(옛 판 「regular + 글」 포함)
|
|
station = {k: v for k, v in station.items() if k not in ("grid", "structure")}
|
|
if station.get("kind") == "irregular":
|
|
station["kind"] = "regular"
|
|
left.append(station)
|
|
kept[key] = station
|
|
kept.update(fresh)
|
|
data["stations"] = sorted(kept.values(), key=lambda s: float(s.get("chainage_m", 0.0)))
|
|
previous = [round(float(station.get("chainage_m", 0.0)), 6) for station in stations]
|
|
summary = _follow_moved_stations(data, before, previous)
|
|
atomic_write_json(path, data)
|
|
_relabel_cross_files(path.parent.parent / "cross_sections", left)
|
|
return summary
|
|
|
|
|
|
async def drop_rows_off_stations(
|
|
connection: aiomysql.Connection, route_id: int, project_root: Path, longitudinal_path: str
|
|
) -> None:
|
|
"""측점을 다시 맞춘 뒤 — 종단 정본에 없는 측점의 횡단 행을 새 측점으로 옮기거나 지운다(272)."""
|
|
path = (project_root.resolve() / longitudinal_path).resolve()
|
|
stations = json.loads(path.read_text(encoding="utf-8")).get("stations") or []
|
|
await sync_cross_section_rows(
|
|
connection, route_id, [float(station["chainage_m"]) for station in stations]
|
|
)
|
|
|
|
|
|
def _relabel_cross_files(cross_dir: Path, stations: list[dict[str, Any]]) -> None:
|
|
"""일반 측점으로 돌아온 자리의 횡단 파일 글(종류 · 구조물)도 맞춘다 — 카드 제목(272)."""
|
|
for station in stations:
|
|
cross_path = find_cross_file(cross_dir, float(station.get("chainage_m", 0.0)))
|
|
if not cross_path.is_file():
|
|
continue
|
|
cross = json.loads(cross_path.read_text(encoding="utf-8"))
|
|
cross.pop("structure", None)
|
|
cross.pop("grid", None)
|
|
cross["kind"] = station.get("kind")
|
|
atomic_write_json(cross_path, cross)
|
|
|
|
|
|
#: 지표 샘플링 조건을 적어 두는 파일 — **나중에 측점을 더 만들 때 같은 조건으로** 뜨기 위해.
|
|
#: ⚠ 조건이 다르면 그 측점만 다른 지표에서 뽑혀 옆 측점과 지반고가 어긋난다.
|
|
SAMPLING_SNAPSHOT_NAME = "sampling.json"
|
|
|
|
|
|
def sampling_snapshot_path(project_root: Path) -> Path:
|
|
return project_root / "B06_Section" / SAMPLING_SNAPSHOT_NAME
|
|
|
|
|
|
def save_sampling_snapshot(project_root: Path, request: RouteConfirmRequest) -> None:
|
|
"""확정 때 쓴 지표 샘플링 조건을 남긴다(2026-09-09).
|
|
|
|
왜 — 관을 나중에 놓으면 그 측점이 안 생기는데(계획서 3-14), 나중에 만들려면 **그때와 같은
|
|
조건**으로 떠야 한다. 조건을 안 남기면 되짚을 길이 없어 **지어내야 하는 자리**가 된다.
|
|
"""
|
|
if not request.filter_key or not request.method:
|
|
return
|
|
path = sampling_snapshot_path(project_root)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(
|
|
path,
|
|
{
|
|
"filter_key": request.filter_key,
|
|
"method": request.method,
|
|
"smooth": bool(request.smooth),
|
|
"surface_model_id": request.surface_model_id,
|
|
},
|
|
)
|
|
|
|
|
|
def load_sampling_snapshot(project_root: Path) -> dict[str, Any] | None:
|
|
"""남겨 둔 샘플링 조건. 없으면 `None` — **지어내지 않는다**(노선 확정을 한 번 더 받는다)."""
|
|
path = sampling_snapshot_path(project_root)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
if not data.get("filter_key") or not data.get("method"):
|
|
return None
|
|
return data
|
|
|
|
|
|
async def _append_irregular_cross_sections(
|
|
connection: aiomysql.Connection,
|
|
project_id: UUID,
|
|
route: dict[str, Any],
|
|
request: RouteConfirmRequest,
|
|
) -> None:
|
|
"""확정 시 비정규 측점의 횡단을 생성해 종단 파일에 병합한다(파일 기반, 비치명적 호출용)."""
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
save_sampling_snapshot(project_root, request)
|
|
stored_options = await get_latest_section_options(connection, project_id)
|
|
crs_epsg = await get_surface_crs_epsg(connection, project_id, request.surface_model_id)
|
|
route_data_path = str(route["route_data_path"])
|
|
|
|
def _generate() -> list[dict[str, Any]]:
|
|
# 측점 목록 = 서버 정본(관 · 구조물)에서 — 화면 목록(관 · A군)을 받아 적으면 서버가 심은
|
|
# 소단 · 옹벽 · 흙막이 · 독립 기슭막이 시작 · 종료 측점이 통째로 빠짐(238-1).
|
|
polyline = _load_route_polyline(project_root, route_data_path)
|
|
extras = resolve_extra_stations(project_root, _load_pipe_points(project_root, polyline))
|
|
return generate_irregular_sections(
|
|
project_root,
|
|
route_data_path,
|
|
request.filter_key,
|
|
request.method,
|
|
request.smooth,
|
|
extra_stations=extras,
|
|
options=_section_options_from_stored(stored_options),
|
|
crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
|
|
)
|
|
|
|
# 빈 목록이어도 맞춤 — 옛 구조물 측점 · 떠난 격자 자리 글을 지워야 함(272)
|
|
irregular_stations = await asyncio.to_thread(_generate)
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route["id"])
|
|
if longitudinal:
|
|
summary = await asyncio.to_thread(
|
|
_merge_irregular_into_longitudinal,
|
|
project_root,
|
|
str(longitudinal["longitudinal_file_path"]),
|
|
irregular_stations,
|
|
)
|
|
if summary is not None: # 계획선을 다시 셈 — DB 등급 요약도 같이(B06 측점 만들기와 같은 길)
|
|
await update_longitudinal_grade_summary(
|
|
connection, route_id=route["id"], grade_summary=summary
|
|
)
|
|
await drop_rows_off_stations(
|
|
connection, route["id"], project_root, str(longitudinal["longitudinal_file_path"])
|
|
)
|