Merge remote-tracking branches 'origin/main_laptop_1', 'origin/sub_laptop_1' and 'origin/sub_laptop_3' into sub_laptop_2

This commit is contained in:
2026-10-21 16:51:10 +09:00
18 changed files with 574 additions and 92 deletions
@@ -54,11 +54,8 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Stream import (
PrimaryRegion,
build_primary_region,
)
from common_util.common_util_route_geometry import (
RouteVertex,
StructureCandidate,
find_stream_crossings,
)
from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
from common_util.common_util_stream_crossing import find_stream_crossings
from config.config_system import (
DRAINAGE_ARROW_BLOCK_M,
DRAINAGE_ARROW_MIN_AGREEMENT,
@@ -47,13 +47,17 @@ from common_util.common_util_drainage_pipes import (
clear_pipe_points,
load_pipe_points,
parse_pipe_points,
pipe_points_path,
pipe_signature,
read_dismissed,
read_pipe_points_file,
route_signature,
save_detail_basins,
save_pipe_points,
)
from common_util.common_util_json import LONLAT_DIGITS, round_floats
from common_util.common_util_route_geometry import StructureCandidate
from common_util.common_util_stream_crossing import keep_labels
from config.config_system import (
DRAINAGE_PIPE_MAX_SPACING_M,
DRAINAGE_PIPE_MIN_SPACING_M,
@@ -611,9 +615,13 @@ async def put_pipe_points(
context, detail, points, _, stale = resolved
signature = route_signature(context.vertices)
# 자동 세류선 관 라벨 되붙임 · 지운 자동 관 = 지움 표지(507-2)
path = pipe_points_path(context.stored_path)
dismissed = read_dismissed(path)
keep_labels(points, read_pipe_points_file(path), dismissed)
# 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
saved = await asyncio.to_thread(
save_pipe_points, context.stored_path, signature, points, context.vertices
save_pipe_points, context.stored_path, signature, points, context.vertices, dismissed
)
if not stale: # 낡은 분석 — 관 자리 · 속성만 정본으로 · 세부 유역 산출물은 안 씀
await asyncio.to_thread(
+29 -6
View File
@@ -171,19 +171,29 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
def followed_pipes(
old_pipes: list, old_xy: list[list[float]], points: list[dict[str, float]], streams: list
old_pipes: list,
old_xy: list[list[float]],
points: list[dict[str, float]],
streams: list,
dismissed: list | None = None,
) -> tuple[list, int, list]:
"""관 따라감 + 새 세류선 교차점 기본 관(저장 안 함) → (짝, 교차 수, 새 노선 정점) — 체인 ·
추천 초기 종단(476) 한 벌."""
추천 초기 종단(476) 한 벌. `dismissed` = 지움 표지(507-2 · 옛 노선 라벨 붙임을 제자리에)."""
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
from B04_PreProcess.B04_PreProcess_Router_Basins import _apply_new_defaults
from common_util.common_util_drainage_pipes import fill_pipe_coordinates
from common_util.common_util_stream_crossing import adopt_labels
vertices = vertices_of([(p["x"], p["y"]) for p in points])
fill_pipe_coordinates(old_pipes, vertices_of(old_xy))
held = [] if dismissed is None else dismissed
if len(old_xy) >= 2: # 라벨 없는 옛 자동 관 · 지움 자리 = 옛 노선 교차 라벨(507-2)
overlap = adopt_labels(old_pipes, held, _base_pipes(vertices_of(old_xy), streams), old_xy)
if overlap:
logger.warning("관 라벨: 세류선 교차 5 m 안 겹친 자동 관 %d개(안 지움)", overlap)
pairs = follow_pipes(old_pipes, vertices)
total = float(vertices[-1].chainage_m)
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total, held)
# 새 교차 관만 B04 와 같은 등록부 기본값(54-39 · 494-2) — 옛 관 빈칸은 미확정 그대로
_apply_new_defaults([pipe for index, pipe in pairs if index < 0])
return pairs, crossing, vertices
@@ -204,6 +214,7 @@ async def _place_pipes(
old_pipes: list,
old_xy: list[list[float]],
points: list[dict[str, float]],
dismissed: list | None = None,
) -> tuple[list, dict[str, int]]:
"""관 따라감 → 새 교차점 기본 관 → 정본 저장(새 노선 지문) · 간격 보충 없음(369).
초기화(492)면 부르는 쪽이 옛 관 · 옛 선을 빈 목록으로 줌 = 새 노선 교차 관만.
@@ -218,8 +229,9 @@ async def _place_pipes(
prepared = await _prepare(project_id)
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음
held = [] if dismissed is None else dismissed
pairs, crossing, vertices = await asyncio.to_thread(
followed_pipes, old_pipes, old_xy, points, streams
followed_pipes, old_pipes, old_xy, points, streams, held
)
await asyncio.to_thread(
save_pipe_points,
@@ -227,6 +239,7 @@ async def _place_pipes(
route_signature(vertices),
[pipe for _, pipe in pairs],
vertices,
held,
)
line = [(p["x"], p["y"]) for p in points]
outside = sum(route_outside(pipe.x, pipe.y, line) for _, pipe in pairs) # 459 · 측점 안 세움
@@ -454,7 +467,11 @@ async def run_route_save_chain(
from B06_Section.B06_Section_Router_Confirm import confirm_sections
from B06_Section.B06_Section_Router_Design import stored_standard_cross_section
from B06_Section.B06_Section_Schema import SectionConfirmRequest
from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file
from common_util.common_util_drainage_pipes import (
pipe_points_path,
read_dismissed,
read_pipe_points_file,
)
from common_util.common_util_initial_snapshot import clear_designing, mark_designing
from common_util.common_util_workflow_state import get_workflow_state
@@ -511,8 +528,14 @@ async def run_route_save_chain(
marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
# 초기화 = 옛 관(옵션 · 사용자 관) 버림 · 새 노선 세류선 교차 관만(492)
# 지움 표지도 — [적용] = 새 초기값(표지 없음 · 507-2)
pairs, pipes = await _place_pipes(
project_id, stored_path, [] if initial else old_pipes, [] if initial else old_xy, points
project_id,
stored_path,
[] if initial else old_pipes,
[] if initial else old_xy,
points,
[] if initial else read_dismissed(pipe_points_path(stored_path)),
)
marks.append(("관 따라감 · 교차점", time.perf_counter()))
+22 -4
View File
@@ -6,7 +6,9 @@ DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름
- 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
(사용자 관 편집 지킴).
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음) ·
같은 세류선 라벨 관(옮겼어도) · 지움 표지가 짝으로 잡은 교차점엔 안 더함(507-2 ·
`common_util_stream_crossing`).
- 관 최대 간격 보충 없음(369) — 간격 보충은 배수유역 [자동 제안]만.
- 측점 열쇠 = `common_util_station_key.station_key` 한 벌(sub2 짝) — 일반 `grid:k` ·
관 `pipe:<id>` · 구조물 `structure:<id>` · 시종점 `end:BP` · `end:EP` ·
@@ -38,6 +40,7 @@ from common_util.common_util_station_key import (
station_key,
station_seat,
)
from common_util.common_util_stream_crossing import claim_crossings, crossing_labels
from common_util.common_util_structure_station import plan_stations, snap_structure_station
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M
@@ -78,18 +81,33 @@ def add_crossing_pipes(
crossings: Sequence[Any],
old_xy: Sequence[Sequence[float]] | None,
total_m: float,
dismissed: Sequence[dict[str, Any]] = (),
) -> int:
"""옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수."""
"""옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관(새 라벨)을 더함 · 더한 수."""
old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
held = [pipe.stream_label for _, pipe in pairs if pipe.stream_label]
claimed = claim_crossings([*held, *(mark for mark in dismissed if "stream" in mark)], crossings)
labels = crossing_labels(crossings)
added = 0
for crossing in crossings:
for index, crossing in enumerate(crossings):
if index in claimed:
continue # 같은 세류선 라벨 관(옮겼어도) · 지운 자동 관(507-2)
if not _moved_away(old_line, crossing.x, crossing.y):
continue # 옛 교차점 — 사용자가 지운 관을 되살리지 않음
at = snap_structure_station(float(crossing.chainage_m), [total_m])
if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs):
continue
pairs.append(
(-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y))
(
-1,
PipePoint(
chainage_m=at,
source=PIPE_SOURCE_STREAM,
x=crossing.x,
y=crossing.y,
stream_label=labels[index],
),
)
)
added += 1
pairs.sort(key=lambda pair: pair[1].chainage_m)
@@ -294,6 +294,11 @@ export function renderProfile(ctx: ProfileRenderContext): void {
),
selectedStationId,
stationDisplay,
mcKeys: new Set(
(longitudinal.stations ?? [])
.filter((station) => station.kind === "curve")
.map((station) => station.chainage_m.toFixed(3)),
),
onCurveRadiusChange: (curve, radius) =>
ctx.applyEdits(setCurveRadius(store.edits(), curve, radius ?? 0)),
// 값 열 계획고 직접 입력 → 규칙 측점과 동일한 station_offset 파이프라인.
+20 -3
View File
@@ -43,6 +43,8 @@ export interface ProfileTableOptions {
irregularStations?: IrregularStation[];
/** 현재 선택된 측점 id. 규칙/비정규 측점 모두 그 측점의 값 열을 테이블에 겹쳐 보여준다. */
selectedStationId?: string | null;
/** MC(곡선 가운데) 측점 누가거리(`toFixed(3)` 열쇠) — 측점 행에 「 (MC)」 · 크게 · 굵게 · 경고색(507-3 덧). */
mcKeys?: ReadonlySet<string>;
/** 이어 공사 시작 기준 — 측점번호·누가거리 표시를 이만큼 더한다(내부 chainage는 0기준 유지). */
stationDisplay?: { station: number; cumulative: number };
onCurveRadiusChange: (curve: AlignmentCurve, radiusM: number | null) => void;
@@ -55,6 +57,8 @@ interface StationRowSpec {
unit: string;
cell: (index: number) => string;
modifier?: string;
/** 측점 이름 행 — MC 열에 (MC) 강조. */
station?: boolean;
}
interface SegmentRowSpec {
@@ -208,6 +212,7 @@ function buildStationRows(
{
label: "측점",
unit: "측점번호+잔여거리",
station: true,
cell: (index) => displayStationLabel(stations[index].chainage_m, interval, display.station),
},
];
@@ -458,6 +463,7 @@ function buildSelectedColumn(
display: { station: number; cumulative: number },
onAdjustStation?: (chainageM: number, deltaM: number) => void,
isIrregular = false,
mc = false,
): HTMLElement {
const plan = interpolateSample(alignment.samples, chainage, "elevation_m");
const ground = interpolateSample(alignment.samples, chainage, "ground_elevation_m");
@@ -487,7 +493,12 @@ function buildSelectedColumn(
{ value: gradeCell((segment) => segment.grade_percent, 2), modifier: "grade" }, // 구배 S
{ value: { text: plan !== null ? plan.toFixed(2) : "" }, modifier: "plan", plan: true }, // 계획고
{ value: { text: ground !== null ? ground.toFixed(2) : "" }, modifier: "" }, // 지반고
{ value: { text: displayStationLabel(chainage, interval, display.station) }, modifier: "" }, // 측점
{
value: {
text: displayStationLabel(chainage, interval, display.station) + (mc ? " (MC)" : ""),
},
modifier: mc ? "mc" : "",
}, // 측점
{ value: { text: curve ? curve.l_m.toFixed(2) : "" }, modifier: "" }, // 곡선 L
{ value: { text: curve ? curve.r_m.toFixed(1) : "" }, modifier: "" }, // 곡선 R
];
@@ -617,10 +628,13 @@ export function createProfileTable(options: ProfileTableOptions): HTMLElement {
alignment.stations.forEach((_, stationIndex) => {
// 구조물 측점 열은 기본값을 비운다 — 격자(빈 셀)만 남기고 값은 선택 오버레이가 맡는다.
const structural = isStructural[stationIndex];
// MC 열 = 측점 행에 「 (MC)」 · 굵게 · 경고색(507-3 덧) — 칸 폭 판정(`longest`)은 옛 글 그대로
const mc =
spec.station && options.mcKeys?.has(alignment.stations[stationIndex].chainage_m.toFixed(3));
const cell = element(
"span",
`b05-profile-table__cell${structural ? " is-structure" : ""}`,
structural ? "" : spec.cell(stationIndex),
`b05-profile-table__cell${structural ? " is-structure" : ""}${mc ? " b06-station--mc" : ""}`,
structural ? "" : spec.cell(stationIndex) + (mc ? " (MC)" : ""),
);
const rotated = stationThickness[stationIndex];
if (rotated !== null) {
@@ -649,6 +663,7 @@ export function createProfileTable(options: ProfileTableOptions): HTMLElement {
// 선택된 측점(규칙·비정규 공용)을 **값 열 오버레이**로 강조한다.
// 규칙 측점은 종점 이동을 반영한 `centers[index]`, 비정규 측점은 `x(chainage)`를 중심으로 쓴다.
const selectedIrregular = selectedIrregularEntry;
const mcAt = (chainageM: number): boolean => options.mcKeys?.has(chainageM.toFixed(3)) ?? false;
const selectedRegularIndex = selectedStationRow
? alignment.stations.indexOf(selectedStationRow)
: -1;
@@ -665,6 +680,7 @@ export function createProfileTable(options: ProfileTableOptions): HTMLElement {
display,
options.onAdjustStation,
true,
mcAt(selectedIrregular.chainage_m),
),
);
} else if (selectedRegularIndex >= 0) {
@@ -678,6 +694,7 @@ export function createProfileTable(options: ProfileTableOptions): HTMLElement {
display,
options.onAdjustStation,
false,
mcAt(alignment.stations[selectedRegularIndex].chainage_m),
),
);
}
@@ -288,6 +288,12 @@
color: rgb(37 99 235);
}
/* MC 측점 이름(507-3 덧) — 굵게 · 경고색 */
.b05-profile-table__irregular-col-cell.is-mc {
color: var(--color-warning);
font-weight: var(--font-weight-bold);
}
/* 계획고(is-plan)는 오버레이에서 별도 강조 없이 다른 값과 동일한 스타일을 쓴다(작업 C-2). */
/* 값 열의 계획고 직접 입력 셀 — 열은 pointer-events:none이라 입력만 되살린다. */
@@ -221,6 +221,7 @@ def _drawing_list(project_root: Path, longitudinal_path: Path) -> list[DesignDra
chainages = sheet["chainages"]
first = station_by_chainage.get(chainages[0], {})
station = station_no_label(float(first.get("chainage_m", chainages[0])), interval)
station += mc_suffix(first)
drawings.append(
DesignDrawingItem(
id=sheet["id"],
@@ -25,7 +25,11 @@ from B05_Profile.B05_Profile_Engine_Sections import (
)
from B05_Profile.B05_Profile_Structures_Repository import load_structures
from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import infer_station_interval, station_no_label
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
infer_station_interval,
mc_suffix,
station_no_label,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_IpTable import clip_lines, ip_table_piece
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Plan import (
_structure_label,
@@ -502,6 +506,11 @@ def build_pieces(
pieces["ip"], pieces["radius"] = curve_pieces(_load_design_curves(project_root), route, box)
if "station" in wanted:
items = []
curve_at = {
float(item["chainage_m"])
for item in longitudinal.get("stations") or []
if mc_suffix(item) and isinstance(item.get("chainage_m"), (int, float))
}
for chainage, x, y in stations:
if not (start_m <= chainage <= end_m):
continue
@@ -509,6 +518,8 @@ def build_pieces(
label = station_no_label(chainage, interval).replace("No.", "NO.")
number_only = "+" not in label
major = number_only and int(label[3:]) % 5 == 0
if chainage in curve_at: # MC = 고칠 자리(507-3) — 5 배수가 아니어도 글 · 큰 글(major)
label, major = label + " (MC)", True
items.append(
{
"chainage_m": chainage,
@@ -178,7 +178,10 @@ export function buildDrawingSidePanel(
}
for (const drawing of items) {
// 횡단면도는 장 단위(여러 측점)라 서버가 준 "N장 (구간)" 라벨을 그대로 쓴다.
section.append(drawingButton(drawing, drawing.label));
const button = drawingButton(drawing, drawing.label);
// 이름 뒤 「 (MC)」(507-3 덧) — 한 칸에 안 들어가 두 열을 가로지름
if (drawing.label.includes("(MC)")) button.dataset.mc = "true";
section.append(button);
}
if (structure && items.length === 0) {
const empty = document.createElement("p");
@@ -70,6 +70,11 @@
grid-column: 1 / -1;
}
/* MC 장 단추(507-3 덧) — 이름이 길어 한 줄 전체 */
.b07-drawing-group[data-kind="cross"] .b07-drawing-button[data-mc="true"] {
grid-column: 1 / -1;
}
.b07-drawing-button {
display: flex;
align-items: center;
+33 -7
View File
@@ -105,6 +105,9 @@ class PipePoint:
# 관 고정 id(347 덧) — 측점 열쇠 `pipe:<id>`(common_util_station_key) · 옮겨도 · 노선이
# 바뀌어도 그대로 · 빈 id 는 `ensure_pipe_ids` 가 읽기 · 되붙이기 · 저장 때 채움.
id: str | None = None
# 자동 세류선 관 라벨(507-2 · `common_util_stream_crossing`) `{stream, order, x, y}` — 옮기고
# 고쳐도 남음 · 화면은 모름(저장 때 id 로 되붙임)
stream_label: dict[str, Any] | None = None
def as_dict(self) -> dict[str, Any]:
# 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
@@ -132,6 +135,8 @@ class PipePoint:
payload["summary_item"] = self.summary_item
if self.group_items:
payload["group_items"] = self.group_items
if self.stream_label:
payload["stream_label"] = self.stream_label
return payload
@@ -437,6 +442,11 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
summary_item=parse_summary_item(item.get("summary_item")),
group_items=parse_group_items(item.get("group_items")),
id=str(item["id"]) if item.get("id") else None,
stream_label=(
dict(item["stream_label"])
if isinstance(item.get("stream_label"), dict)
else None
),
)
)
points.sort(key=lambda point: point.chainage_m)
@@ -500,9 +510,12 @@ def save_pipe_points(
signature: str,
points: list[PipePoint],
vertices: list[RouteVertex] | None = None,
dismissed: list[dict[str, Any]] | None = None,
) -> int:
"""관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다.
`dismissed` = 지운 자동 관 지움 표지(507-2) — None 이면 파일의 것 그대로.
`vertices`를 주면 좌표가 빈 관을 그 노선 위 좌표로 채워 둔다 — 다음에 다른 선으로
읽어도 그 자리에 되놓을 수 있다. 좌표가 그 누가거리를 안 낳으면(누가거리만 옮긴 낡은 좌표)
노선 위로 · 낳으면 땅 위 자리 그대로(493-4 · 구조물 `fill_structure_coordinates` · TS
@@ -516,17 +529,29 @@ def save_pipe_points(
ensure_pipe_ids(points)
path = pipe_points_path(stored_path)
path.parent.mkdir(parents=True, exist_ok=True)
atomic_write_json(
path,
{
"route_signature": signature,
"points": [point.as_dict() for point in points],
},
)
document: dict[str, Any] = {
"route_signature": signature,
"points": [point.as_dict() for point in points],
}
marks = read_dismissed(path) if dismissed is None else dismissed
if marks:
document["dismissed_streams"] = marks
atomic_write_json(path, document)
logger.info("배수유역: 관 지점 %d개를 저장했습니다 (%s).", len(points), path.name)
return len(points)
def read_dismissed(path: Path) -> list[dict[str, Any]]:
"""지운 자동 관 지움 표지(507-2 `dismissed_streams`) — 없거나 깨졌으면 빈 목록."""
try:
marks = json.loads(path.read_text(encoding="utf-8")).get("dismissed_streams")
except (OSError, ValueError, AttributeError):
return []
return (
[dict(mark) for mark in marks if isinstance(mark, dict)] if isinstance(marks, list) else []
)
def read_pipe_points_file(path: Path) -> list[PipePoint]:
"""저장분을 **노선 지문과 무관하게** 그대로 읽는다. 파일이 없거나 깨졌으면 빈 목록.
@@ -564,6 +589,7 @@ def append_pipe_points_file(path: Path, points: list[PipePoint]) -> int | None:
atomic_write_json(
path,
{
**document, # 지움 표지(507-2) 그대로
"route_signature": str(document.get("route_signature") or ""),
"points": [*stored, *(point.as_dict() for point in points)],
},
+4 -63
View File
@@ -1,4 +1,4 @@
"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점.
"""계획 노선 기하 공용 유틸 — 정점·누가거리(세류 교차점 = `common_util_stream_crossing`).
배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만
정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다.
@@ -19,7 +19,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
from shapely.geometry import LineString, Point, shape
from shapely.geometry import LineString
from config.config_system import SURFACE_ROUTE_EDGE_TRIM_M
@@ -53,6 +53,8 @@ class StructureCandidate:
# "stream"=세류 교차, "spacing"=최대 간격 규칙 보충, "confirmed"=사용자 확정
reason: str
stream_name: str | None = None
# 세류선 지물 id(UFID · 507-2 관 라벨) — 없으면 None
stream_id: str | None = None
@dataclass
@@ -623,67 +625,6 @@ def is_uphill_at(vertices: list[RouteVertex], chainage_m: float, window_m: float
return forward_z >= back_z
def find_stream_crossings(
vertices: list[RouteVertex],
stream_features: list[dict[str, Any]],
) -> list[StructureCandidate]:
"""노선 평면 선형과 세류선의 교차 지점을 누가거리 순으로 찾는다."""
if len(vertices) < 2:
return []
route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
candidates: list[StructureCandidate] = []
for feature in stream_features:
geometry = feature.get("geometry")
if not geometry:
continue
try:
stream = shape(geometry)
except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다
continue
if stream.is_empty:
continue
intersection = route_line.intersection(stream)
if intersection.is_empty:
continue
name = _stream_name(feature)
for point in _collect_points(intersection):
candidates.append(
StructureCandidate(
chainage_m=route_line.project(point),
x=point.x,
y=point.y,
reason="stream",
stream_name=name,
)
)
candidates.sort(key=lambda item: item.chainage_m)
return candidates
def _stream_name(feature: dict[str, Any]) -> str | None:
properties = feature.get("properties") or {}
for key in ("명칭", "하천명", "NAME", "name"):
value = properties.get(key)
if value:
return str(value)
return None
def _collect_points(geometry: Any) -> list[Point]:
"""교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다."""
if geometry.geom_type == "Point":
return [geometry]
if geometry.geom_type in {"MultiPoint", "GeometryCollection"}:
points: list[Point] = []
for part in geometry.geoms:
points.extend(_collect_points(part))
return points
# 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다.
if geometry.geom_type in {"LineString", "MultiLineString"}:
return [geometry.interpolate(0.5, normalized=True)]
return []
def _first_text(row: dict[str, Any], keys: tuple[str, ...]) -> str | None:
for key in keys:
value = row.get(key)
+222
View File
@@ -0,0 +1,222 @@
"""세류선 교차점 · 자동 세류선 관 라벨(507-2).
교차점 = 노선 평면 선형 × 세류선(`find_stream_crossings`) — 지물 id(`stream_id`)를 함께 단다.
자동 세류선 관 라벨(`PipePoint.stream_label`) = `{stream, order, x, y}` — 세류선 지물 id · 그
세류선의 노선 위 교차 차례 · 처음 교차점 좌표. 끌거나 옵션을 고쳐 `source` 가 바뀌어도 남음.
- 노선 바뀐 [저장](`B05_Profile_Route_Save.followed_pipes`) — 라벨(관 · 지움 표지)마다 같은
세류선의 새 교차점 중 라벨 좌표와 가장 가까운 것을 짝으로 잡음 · 짝 잡힌 교차점엔 관 안 더함.
- 지움 표지(`pipe_points.json` `dismissed_streams`) = 사용자가 지운 자동 관 라벨 — 다시 안 생김.
라벨 없는 자동 관을 지우면 그 자리(x · y)만 남겨 두고 다음 노선 [저장] 때 라벨로 바꿈.
- 라벨 없는 옛 관(`source` stream) = 노선 [저장] 때 옛 노선 교차점 5 m 안이면 그 교차 라벨 ·
교차 하나에 관 하나(가장 가까운 관) · 나머지 겹친 관은 지우지 않고 수만 셈.
- 추천 [적용] · [초기화] = 초기값 파일 그대로(지움 표지도 초기값).
"""
from __future__ import annotations
import math
from typing import Any, Sequence
from shapely.geometry import LineString, Point, shape
from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M
def stream_key(feature: dict[str, Any], index: int) -> str:
"""세류선 지물 id — 도엽 UFID · 없으면 피처 id · 그것도 없으면 파일 안 차례."""
properties = feature.get("properties") or {}
value = properties.get("UFID") or feature.get("id")
return str(value) if value not in (None, "") else f"#{index}"
def find_stream_crossings(
vertices: list[RouteVertex],
stream_features: list[dict[str, Any]],
) -> list[StructureCandidate]:
"""노선 평면 선형과 세류선의 교차 지점을 누가거리 순으로 찾는다."""
if len(vertices) < 2:
return []
route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
candidates: list[StructureCandidate] = []
for index, feature in enumerate(stream_features):
geometry = feature.get("geometry")
if not geometry:
continue
try:
stream = shape(geometry)
except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다
continue
if stream.is_empty:
continue
intersection = route_line.intersection(stream)
if intersection.is_empty:
continue
name = _stream_name(feature)
for point in _collect_points(intersection):
candidates.append(
StructureCandidate(
chainage_m=route_line.project(point),
x=point.x,
y=point.y,
reason="stream",
stream_name=name,
stream_id=stream_key(feature, index),
)
)
candidates.sort(key=lambda item: item.chainage_m)
return candidates
def _stream_name(feature: dict[str, Any]) -> str | None:
properties = feature.get("properties") or {}
for key in ("명칭", "하천명", "NAME", "name"):
value = properties.get(key)
if value:
return str(value)
return None
def _collect_points(geometry: Any) -> list[Point]:
"""교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다."""
if geometry.geom_type == "Point":
return [geometry]
if geometry.geom_type in {"MultiPoint", "GeometryCollection"}:
points: list[Point] = []
for part in geometry.geoms:
points.extend(_collect_points(part))
return points
# 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다.
if geometry.geom_type in {"LineString", "MultiLineString"}:
return [geometry.interpolate(0.5, normalized=True)]
return []
# ── 자동 세류선 관 라벨(507-2) ────────────────────────────────────────────────
def crossing_labels(crossings: Sequence[Any]) -> list[dict[str, Any]]:
"""교차점(누가거리 순)마다 라벨 — 차례 = 같은 세류선 안에서 몇 번째 교차인가."""
seen: dict[Any, int] = {}
labels = []
for crossing in crossings:
stream = getattr(crossing, "stream_id", None)
seen[stream] = seen.get(stream, 0) + 1
labels.append(
{
"stream": stream,
"order": seen[stream],
"x": round(float(crossing.x), 3),
"y": round(float(crossing.y), 3),
}
)
return labels
def _gap(label: dict[str, Any], x: float, y: float) -> float:
return math.hypot(float(label["x"]) - x, float(label["y"]) - y)
def claim_crossings(labels: Sequence[dict[str, Any]], crossings: Sequence[Any]) -> set[int]:
"""라벨마다 같은 세류선 교차점 중 라벨 좌표와 가장 가까운 것 하나 — 잡힌 교차점 번호.
가까운 짝부터 하나씩(라벨 하나 · 교차점 하나) · 거리 한계 없음.
"""
pairs = sorted(
(_gap(label, float(crossing.x), float(crossing.y)), li, ci)
for li, label in enumerate(labels)
if "stream" in label
for ci, crossing in enumerate(crossings)
if getattr(crossing, "stream_id", None) == label["stream"]
)
used_labels: set[int] = set()
used: set[int] = set()
for _, li, ci in pairs:
if li in used_labels or ci in used:
continue
used_labels.add(li)
used.add(ci)
return used
def adopt_labels(
pipes: Sequence[Any],
dismissed: list[dict[str, Any]],
crossings: Sequence[Any],
old_xy: Sequence[Sequence[float]],
) -> int:
"""옛 노선 교차점 중 라벨(관 · 지움 표지)이 안 잡은 것 — 옛 노선 누가거리 5 m 안 라벨 없는
stream 관(가장 가까운 것) · 없으면 자리만 남은 지움 표지에 그 교차 라벨(제자리) · 돌려줌 =
겹쳐 라벨 못 받은 stream 관 수.
거리 = 옛 노선에 투영한 누가거리 차(옛 관 좌표가 다른 선 위에 남은 저장분 · 옆 15 m 실측).
"""
from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM
line = LineString(old_xy)
labels = crossing_labels(crossings)
held = [pipe.stream_label for pipe in pipes if pipe.stream_label] + [
mark for mark in dismissed if "stream" in mark
]
claimed = claim_crossings(held, crossings)
loose = [
(line.project(Point(pipe.x, pipe.y)), pipe)
for pipe in pipes
if not pipe.stream_label and pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None
]
spots = [
(line.project(Point(float(mark["x"]), float(mark["y"]))), mark)
for mark in dismissed
if "stream" not in mark
]
def nearest(items: list, at: float) -> int | None:
gaps = [(abs(chainage - at), order) for order, (chainage, _) in enumerate(items)]
best = min(gaps, default=None)
return best[1] if best and best[0] <= DRAINAGE_PIPE_MIN_SPACING_M else None
for index, (label, crossing) in enumerate(zip(labels, crossings)):
if index in claimed:
continue
at = float(crossing.chainage_m)
pick = nearest(loose, at)
if pick is not None:
loose.pop(pick)[1].stream_label = dict(label)
continue
pick = nearest(spots, at)
if pick is not None:
spots.pop(pick)[1].update(label)
near = [float(crossing.chainage_m) for crossing in crossings]
return sum(
1 for chainage, _ in loose if nearest([(c, None) for c in near], chainage) is not None
)
def keep_labels(
points: Sequence[Any], stored: Sequence[Any], dismissed: list[dict[str, Any]]
) -> None:
"""관 저장(PUT) — 화면이 모르는 라벨을 정본에서 id 로 되붙임 · 지운 자동 관 = 지움 표지(제자리).
라벨 있는 관 = 라벨 · 라벨 없는 stream 관 = 그 자리(x · y · 다음 노선 [저장] 때 라벨로).
"""
from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM
by_id = {pipe.id: pipe for pipe in stored if pipe.id}
for point in points:
old = by_id.get(point.id)
if not point.stream_label and old is not None and old.stream_label:
point.stream_label = dict(old.stream_label)
kept = {point.id for point in points}
for pipe in stored:
if pipe.id in kept:
continue
if pipe.stream_label:
mark = dict(pipe.stream_label)
elif pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None and pipe.y is not None:
mark = {"x": round(float(pipe.x), 3), "y": round(float(pipe.y), 3)}
else:
continue
if mark not in dismissed:
dismissed.append(mark)
@@ -30,3 +30,10 @@
## 요구 313 (10-21 · 사용자 · 311-2 고침)
1. 노출형 횡단수로와 개거에 대해 잘 못 지시함. 다른 구조물 중 취수정이나 국가지점 번호 처럼 동작해야함. 되돌리면 될것같음. 확인
## 요구 314 (10-21 · 사용자 · 508 다음 장 후보에 답)
1. 시험 16개 정검 및 도우미 정리는 진행
2. b05 b06의 자동관표시는 추천 노천의 미리뵉 모달창에 띄운다는 말인지? 추천 다시셈이라는게 전처리에 저장된값을 다시걔산한다는 말임?
3. b07의 횡단도가 도각이나 칸 밖으로나가는건 사용자가 조절해야하는 부분임.
4. 계산자리의 제어페이지를 만드는 이유는 프론트와 백엔드에서 일원화된 동작을 하게하고 테스트하면서 사용자편의성을 보고 선택하기 위함임. 이해함?
@@ -81,7 +81,9 @@ def test_추천_적용_체인_관_저장(monkeypatch):
monkeypatch.setattr(watershed, "_prepare", prepare)
monkeypatch.setattr(
drainage_pipes, "save_pipe_points", lambda path, sig, pipes, verts: written.extend(pipes)
drainage_pipes,
"save_pipe_points",
lambda path, sig, pipes, verts, *_: written.extend(pipes),
)
pairs, counts = asyncio.run(save._place_pipes(None, "unused", [], [], _points(NEW)))
assert counts["crossing"] == 1
@@ -0,0 +1,154 @@
# -*- coding: utf-8 -*-
"""507-2 자동 세류선 관 라벨 — 노선 바꾼 [저장] 에 관이 여럿 생기던 흠(요구 311-3).
못박는 것
- 라벨 = 세류선 지물 id · 교차 차례 · 처음 교차점 좌표 · 끌어 source 가 user 로 바뀌어도 남음 ·
파일 · 화면 PUT(라벨 모름)을 지나도 id 로 되붙음.
- 노선 바꾼 [저장] 두 번 = 같은 세류선 라벨 관이 있으면(옮겼어도) 안 더함 · 관 수 그대로.
- 지운 자동 관 = 지움 표지 · 다시 안 생김(라벨 없던 옛 관도 자리로 남겨 다음 [저장] 때 라벨로).
- 초기값([적용]) = 지움 표지 없음 · 새 교차 관 · save_pipe_points 표지 None = 파일 것 그대로.
- 라벨 없는 옛 stream 관 = 옛 노선 교차 5 m 안이면 라벨 · 겹친 관은 안 지우고 수만 셈.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import B05_Profile.B05_Profile_Route_Save as save # noqa: E402
import common_util.common_util_drainage_pipes as drainage_pipes # noqa: E402
from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of # noqa: E402
from common_util.common_util_drainage_pipes import ( # noqa: E402
PIPE_SOURCE_STREAM,
PipePoint,
parse_pipe_points,
)
from common_util.common_util_stream_crossing import ( # noqa: E402
adopt_labels,
find_stream_crossings,
keep_labels,
)
OLD = [[0.0, 0.0], [100.0, 0.0], [200.0, 0.0], [300.0, 0.0], [400.0, 0.0]]
# 앞 측점 노선을 북으로 — 세류선 교차가 30 m · 45 m 옮음(5 m 솎기 넘음)
NEW1 = [[0.0, 0.0], [100.0, 0.0], [200.0, 60.0], [300.0, 0.0], [400.0, 0.0]]
NEW2 = [[0.0, 0.0], [100.0, 0.0], [200.0, 90.0], [300.0, 0.0], [400.0, 0.0]]
STREAMS = [
{
"type": "Feature",
"properties": {"UFID": "S1", "구분": "세류"},
"geometry": {"type": "LineString", "coordinates": [[150.0, -50.0], [150.0, 120.0]]},
}
]
def _points(xy) -> list[dict[str, float]]:
return [{"x": x, "y": y} for x, y in xy]
def _save(old: list[PipePoint], old_xy, new_xy, dismissed=None) -> tuple[list[PipePoint], int]:
pairs, crossing, _ = save.followed_pipes(old, old_xy, _points(new_xy), STREAMS, dismissed)
return [pipe for _, pipe in pairs], crossing
def _legacy() -> PipePoint:
"""라벨 없는 옛 자동 관(옛 노선 교차 (150, 0))."""
return PipePoint(chainage_m=150.0, source=PIPE_SOURCE_STREAM, x=150.0, y=0.0, id="p15000")
def test_교차점에_세류선_id():
found = find_stream_crossings(vertices_of(OLD), STREAMS)
assert [(c.stream_id, round(c.x), round(c.y)) for c in found] == [("S1", 150, 0)]
def test_노선_바꾼_저장_두_번_관_수_그대로():
pipes, crossing = _save([_legacy()], OLD, NEW1)
assert crossing == 0 and len(pipes) == 1 # 고치기 전 = 옛 관 + 새 관 둘
label = pipes[0].stream_label
assert label == {"stream": "S1", "order": 1, "x": 150.0, "y": 0.0} # 처음 교차점
# 파일을 지나도 라벨 그대로 · 두 번째 노선 [저장]
again = parse_pipe_points([pipe.as_dict() for pipe in pipes])
pipes2, crossing2 = _save(again, NEW1, NEW2)
assert crossing2 == 0 and len(pipes2) == 1 and pipes2[0].stream_label == label
def test_끌어_user_로_바뀌어도_라벨_남음():
pipes, _ = _save([_legacy()], OLD, NEW1)
moved = pipes[0]
moved.source, moved.x, moved.y = "user", 170.0, 40.0 # 끌기 = user
pipes2, crossing2 = _save(parse_pipe_points([moved.as_dict()]), NEW1, NEW2)
assert crossing2 == 0 and len(pipes2) == 1 and pipes2[0].stream_label["stream"] == "S1"
def test_화면_PUT_라벨_되붙임():
stored = [
PipePoint(chainage_m=150.0, id="p15000", stream_label={"stream": "S1", "x": 1, "y": 2})
]
request = [PipePoint(chainage_m=152.0, id="p15000", source="user")] # 화면은 라벨 모름
dismissed: list = []
keep_labels(request, stored, dismissed)
assert request[0].stream_label == stored[0].stream_label and dismissed == []
def test_지운_자동_관_다시_안_생김():
pipes, _ = _save([_legacy()], OLD, NEW1)
dismissed: list = []
keep_labels([], pipes, dismissed) # 사용자가 지우고 [저장]
assert dismissed == [pipes[0].stream_label]
after, crossing = _save([], NEW1, NEW2, dismissed)
assert crossing == 0 and after == []
def test_라벨_없던_옛_관_지움도_다시_안_생김():
dismissed: list = []
keep_labels([], [_legacy()], dismissed)
assert dismissed == [{"x": 150.0, "y": 0.0}] # 자리만
after, crossing = _save([], OLD, NEW1, dismissed)
assert crossing == 0 and after == []
assert dismissed[0]["stream"] == "S1" # 다음 [저장] 때 라벨로
def test_사용자_관_지움은_표지_없음():
dismissed: list = []
keep_labels([], [PipePoint(chainage_m=50.0, x=50.0, y=0.0)], dismissed)
assert dismissed == []
def test_초기값_적용_새_교차_관():
pipes, crossing = _save([], [], NEW1, [])
assert crossing == 1 and pipes[0].stream_label["stream"] == "S1"
assert pipes[0].stream_label["y"] == 30.0
def test_지움_표지_파일(monkeypatch, tmp_path):
path = tmp_path / "pipe_points.json"
monkeypatch.setattr(drainage_pipes, "pipe_points_path", lambda _stored: path)
mark = [{"stream": "S1", "order": 1, "x": 150.0, "y": 0.0}]
drainage_pipes.save_pipe_points("x", "sig", [], None, mark)
drainage_pipes.save_pipe_points("x", "sig", [PipePoint(chainage_m=5.0)]) # None = 그대로
assert json.loads(path.read_text("utf-8"))["dismissed_streams"] == mark
assert drainage_pipes.read_dismissed(path) == mark
drainage_pipes.save_pipe_points("x", "sig", [], None, []) # [적용] = 표지 없음
assert "dismissed_streams" not in json.loads(path.read_text("utf-8"))
def test_겹친_옛_관_안_지우고_셈():
twin = PipePoint(chainage_m=152.0, source=PIPE_SOURCE_STREAM, x=152.0, y=0.0, id="p15200")
old = [_legacy(), twin]
overlap = adopt_labels(old, [], find_stream_crossings(vertices_of(OLD), STREAMS), OLD)
assert overlap == 1
assert old[0].stream_label and not old[1].stream_label # 가까운 관만 라벨
pipes, crossing = _save(parse_pipe_points([p.as_dict() for p in old]), OLD, NEW1)
assert crossing == 0 and len(pipes) == 2 # 둘 다 그대로 · 새 관 없음
def test_옆_다른_선_위_옛_관도_라벨():
"""옛 관 좌표가 다른 선 위(옆 15 m · 검증_sub3 실측) — 옛 노선 누가거리로 5 m 안이면 라벨."""
side = PipePoint(chainage_m=150.0, source=PIPE_SOURCE_STREAM, x=151.0, y=15.0, id="p15000")
pipes, crossing = _save([side], OLD, NEW1)
assert crossing == 0 and len(pipes) == 1 and pipes[0].stream_label["stream"] == "S1"
+36
View File
@@ -87,3 +87,39 @@ def test_b05_b06_screen_mc_emphasis() -> None:
assert 'themeColor("--color-warning"' in ticks
swap = _src("B06_Section/logic/B06_Section_Station_Swap.ts")
assert 'formatStation(station.chainageM) + (kind === "curve" ? " (MC)" : "")' in swap
def test_b07_plan_station_mc_label() -> None:
"""계획평면도 측점 글 = 5 배수만이었음 → MC 도 큰 글(major) + (MC)."""
plan = _src("B07_DesignDetail/B07_DesignDetail_Router_Support_Plan.py")
assert 'label, major = label + " (MC)", True' in plan
assert "if mc_suffix(item)" in plan # 종단 kind curve 측점만
def test_b07_sheet_button_mc() -> None:
"""B07 횡단 장 단추 「3장 (No.29+19.0 (MC))」 — 그 장 첫 측점이 MC 일 때."""
support = _src("B07_DesignDetail/B07_DesignDetail_Router_Support.py")
assert "station += mc_suffix(first)" in support
assert "label=f\"{sheet['number']}장 ({station})\"" in support
def test_b05_profile_table_station_mc() -> None:
table = _src("B05_Profile/B05_Profile_UI_Profile_Table.ts")
# 칸 글 · 굵게 · 경고색 = B06 공용 규칙 · 칸 폭 판정(`longest`)은 (MC) 없는 글 그대로
assert 'spec.cell(stationIndex) + (mc ? " (MC)" : "")' in table
assert '${mc ? " b06-station--mc" : ""}' in table
assert 'display.station) + (mc ? " (MC)" : "")' in table # 고른 측점 값 열
render = _src("B05_Profile/B05_Profile_UI_Profile_Render.ts")
assert 'station.kind === "curve"' in render
css = _src("B05_Profile/B05_Profile_UI_Style_Table.css")
body = css[css.index("__irregular-col-cell.is-mc {") : css.index("}", css.index(".is-mc {"))]
assert "var(--color-warning)" in body
assert "var(--font-weight-bold)" in body
def test_b07_sheet_button_spans_row() -> None:
panels = _src("B07_DesignDetail/B07_DesignDetail_UI_Panels.ts")
assert 'button.dataset.mc = "true"' in panels
css = _src("B07_DesignDetail/B07_DesignDetail_UI_Style.css")
rule = css[css.index('.b07-drawing-button[data-mc="true"] {') :]
assert "grid-column: 1 / -1" in rule[: rule.index("}")]