Merge remote-tracking branches 'origin/main_desktop_1' and 'origin/sub_laptop_6' into sub_laptop_5

This commit is contained in:
2026-09-29 10:15:36 +09:00
5 changed files with 370 additions and 18 deletions
@@ -341,17 +341,34 @@ def generate_sections(
if 0.0 <= float(chainage) <= total
}
if extras:
# 횡단 파일명이 정수 미터 반올림이라(cross_filename) 1m 안의 두 측점은 같은 파일을
# 덮어써 측점 수와 횡단 파일 수가 어긋난다. 같은 정수 미터면 격자 측점으로 스냅해
# 라벨만 얹고(아래 station["structure"] 부착), 비정규끼리 겹치면 앞선 것만 남긴다.
# ponytail: 파일명이 정수 미터인 동안의 가드. 소수 1자리 규칙으로 바꾸면 지운다.
buckets = {int(round(float(value))): round(float(value), 6) for value in station_chainage}
# 횡단 파일명이 정수 미터 반올림이라(cross_filename) 한 정수 미터에 측점은 하나만 둔다.
# 관 · 구조물 자체 측점은 소수 그대로 — 격자 측점과 같은 정수 미터면 격자 대신 그 측점을
# 쓴다(시 · 종점 격자만 그대로 두고 라벨을 얹음). 비정규끼리 겹치면 앞선 것만 남긴다.
# ponytail: 파일명이 정수 미터인 동안의 가드. 소수 이름 규칙으로 바꾸면 지운다.
grid = [round(float(value), 6) for value in station_chainage]
ends = {grid[0], grid[-1]}
buckets = {int(round(value)): value for value in grid}
taken: set[int] = set()
dropped: set[float] = set()
snapped: dict[float, str] = {}
for chainage in sorted(extras):
snapped.setdefault(buckets.setdefault(int(round(chainage)), chainage), extras[chainage])
bucket = int(round(chainage))
if bucket in taken:
continue
taken.add(bucket)
regular = buckets.get(bucket)
if regular in ends:
snapped[regular] = extras[chainage]
continue
if regular is not None:
dropped.add(regular)
snapped[chainage] = extras[chainage]
extras = snapped
station_chainage = np.unique(
np.r_[station_chainage, np.array(sorted(extras), dtype=np.float64)]
np.r_[
np.array([value for value in grid if value not in dropped], dtype=np.float64),
np.array(sorted(extras), dtype=np.float64),
]
)
station_xy = _interpolate_xy(points, route_chainage, station_chainage)
tangents = np.vstack(
@@ -0,0 +1,86 @@
/* 초기노선 갈래 **저장 왕복**(계획서 20) 헬퍼 — 서버 답을 화면 코드 그대로 편다.
*
* 표준입력 JSON `{ jobs: [...] }` 를 받아 일마다 결과 한 줄을 돌려준다.
* - `apply` : `/route/initial` 답 → 편집값(`normalizeRouteInitial`) → [확인] 몸통
* (`_RouteEdit_Apply` 의 vertices 만들기 · `_Initial.current()` 의 initial)
* - `reopen` : `/route/plan` 답 → 다시 연 노드(`flattenServerPlan`) · 현재 초기값(`serverBase`)
* · 캐시 초기값과 같은가(`sameBase` — [초기화]가 캐시 노드를 쓸지 서버에 물을지)
*
* ⚠ 확장자가 `.cjs` 인 까닭은 `helper_b05_cut_merged_loop.cjs` 머리에 적힌 것과 같다.
*/
const fs = require("fs");
const path = require("path");
const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript"));
const ROOT = path.join(__dirname, "..", "..");
const cache = new Map();
function memoryStorage() {
const map = new Map();
return {
getItem: (key) => (map.has(key) ? map.get(key) : null),
setItem: (key, value) => map.set(key, String(value)),
removeItem: (key) => map.delete(key),
key: (index) => [...map.keys()][index] ?? null,
get length() {
return map.size;
},
};
}
globalThis.window = { localStorage: memoryStorage(), sessionStorage: memoryStorage() };
globalThis.fetch = () => Promise.resolve({ ok: false });
const STUBS = { "@ui/ui_template_palette": { themeColor: (_name, fallback) => fallback } };
function loadTs(file) {
const full = path.resolve(file);
if (cache.has(full)) return cache.get(full);
const js = ts.transpileModule(fs.readFileSync(full, "utf8"), {
compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
}).outputText;
const box = { exports: {} };
cache.set(full, box.exports);
const localRequire = (request) => {
if (STUBS[request]) return STUBS[request];
if (request.startsWith(".")) return loadTs(path.join(path.dirname(full), `${request}.ts`));
return require(request);
};
new Function("exports", "module", "require", js)(box.exports, box, localRequire);
cache.set(full, box.exports);
return box.exports;
}
const model = loadTs(path.join(ROOT, "B05_Profile", "B05_Profile_UI_RouteEdit_Initial_Model.ts"));
const edits = loadTs(path.join(ROOT, "B05_Profile", "B05_Profile_UI_RouteEdit_Edits.ts"));
const { jobs } = JSON.parse(fs.readFileSync(0, "utf8"));
const out = jobs.map((job) => {
if (job.kind === "apply") {
const choice = model.normalizeRouteInitial(job.response, job.logic, job.weights ?? null);
const { planned, curveOn, curveRadius } = choice.snapshot;
return {
choice,
// `_RouteEdit_Apply` [확인] 몸통 그대로.
body: {
vertices: planned.map(([x, y], index) => ({
x,
y,
curve: curveOn[index] !== false,
radius_m: curveRadius[index] ?? null,
})),
initial: { mode: choice.logic, weights: choice.weights },
},
label: model.choiceLabel(choice.logic, choice.weights),
metrics: model.metricsLine(choice.metrics),
};
}
const flat = edits.flattenServerPlan(job.plan.nodes ?? [], job.plan.curves ?? []);
const base = model.serverBase(job.plan.current_initial);
return {
planned: flat.planned,
base,
label: model.choiceLabel(base.logic, base.weights),
sameAsCache: job.cached ? model.sameBase(job.cached, base) : null,
};
});
process.stdout.write(JSON.stringify(out));
@@ -192,7 +192,7 @@ def test_matching_signature_loads_points(tmp_path):
]
# ── 횡단 파일명 충돌 스냅(격자 우선 불변식) ────────────────────────────────
# ── 횡단 파일명 충돌 — 한 정수 미터에 측점 하나 · 관 · 구조물 자체 측점 우선(PLAN 19장) ──
def _stations(extra_stations):
@@ -204,17 +204,31 @@ def _stations(extra_stations):
return result["longitudinal"]["stations"]
def test_extra_station_within_one_meter_snaps_to_grid():
def test_extra_station_in_grid_meter_replaces_grid_station():
stations = _stations(((100.4, "X"),))
assert 100.4 not in [station["chainage_m"] for station in stations]
chainages = [station["chainage_m"] for station in stations]
# 관 · 구조물 자체 측점 그대로 — 같은 정수 미터 격자 100 은 빠져 파일(cross_00100m)은 한 장
assert 100.4 in chainages and 100.0 not in chainages
labelled = [station for station in stations if station.get("structure") == "X"]
assert [station["chainage_m"] for station in labelled] == [100.0]
assert labelled[0]["kind"] == "regular" # 격자 측점 우선 — 기존 불변식
assert [(station["chainage_m"], station["kind"]) for station in labelled] == [
(100.4, "irregular")
]
assert len({cross_filename(c) for c in chainages}) == len(chainages)
def test_extra_station_at_route_end_keeps_end_station():
stations = _stations(((0.3, "S"), (349.8, "E")))
chainages = [station["chainage_m"] for station in stations]
# 시 · 종점은 격자 그대로 두고 라벨만 얹음
assert chainages[0] == 0.0 and chainages[-1] == 350.0
assert 0.3 not in chainages and 349.8 not in chainages
assert stations[0].get("structure") == "S" and stations[-1].get("structure") == "E"
def test_two_extras_in_same_meter_bucket_keep_one():
stations = _stations(((80.2, "A"), (80.4, "B")))
assert [station["chainage_m"] for station in stations if station.get("structure")] == [80.0]
assert [station["chainage_m"] for station in stations if station.get("structure")] == [80.2]
assert 80.0 not in [station["chainage_m"] for station in stations]
def test_live_extras_stay_irregular_and_filenames_stay_unique():
@@ -0,0 +1,235 @@
"""초기노선 갈래 저장 왕복 — [확인]에 실은 초기값이 다시 열기 뒤 그대로인가(계획서 20).
환경 배정 storage 프로젝트(데스크탑_메인)의 **사본**에서 화면과 같은 차례로 서버 길을 부른다.
갈래(추종 · 토공 최소 · 기울기 순한 · 비중 50/25/25)마다
① `POST /route/initial` → 비교값이 다 있음
② 화면 코드로 [확인] 몸통을 만들어 `POST /route/replan` → 서버가 initial 을 적음
③ `GET /route/plan` → 현재 초기값 = 보낸 것 · 다시 연 노드 = 보낸 노드
④ [초기화] — 캐시가 있으면 그 노드, 없으면(다른 PC) 서버가 같은 갈래로 같은 노드를 냄
화면 쪽 셈은 `helper_b05_routeedit_initial_apply.cjs` 가 화면 코드 그대로 돌린다.
- 파일: 프로젝트 폴더를 임시 폴더로 복사(원본 입력 · 점군 · 큰 파일 제외) ·
`STORAGE_BASE_DIR` 를 그리로.
- DB: 읽기 전용 세션으로 진짜 DB 를 읽기만 함.
- 재계산 체인(`_recompute`, 배수유역부터 약 90초 · DB 에 씀)은 성공만 돌려주는 가짜로 갈음 —
여기서 보는 것은 초기값이 적히고 다시 읽히는 길이다.
- 원본 프로젝트 · 사용자 프로젝트 · DB 는 바뀌지 않음.
"""
from __future__ import annotations
import json
import shutil
import subprocess
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any
import pytest
ROOT = Path(__file__).resolve().parents[2]
PROJECT_ID = "79da822d-921a-40f8-bb93-d8ffd90eeedf" # OWNERS.md — desktop-main `데스크탑_메인`
SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID
_COPY_DIRS = ("B04_PreProcess/drainage", "B05_Profile", "B06_Section", "initial_snapshot")
_COPY_MAX_BYTES = 20 * 1024 * 1024
HELPER = Path(__file__).with_name("helper_b05_routeedit_initial_apply.cjs")
pytestmark = pytest.mark.skipif(not SOURCE.is_dir(), reason="배정 storage 프로젝트가 없음")
LOGICS: list[tuple[str, dict[str, float] | None]] = [
("follow", None),
("earthwork", None),
("gentle", None),
("weighted", {"expected": 50, "earthwork": 25, "grade": 25}),
]
def _copy_project(dest: Path) -> None:
for rel in _COPY_DIRS:
shutil.copytree(SOURCE / rel, dest / rel)
for rel in ("B04_PreProcess/models", "B04_PreProcess/processed"):
(dest / rel).mkdir(parents=True, exist_ok=True)
for item in (SOURCE / rel).iterdir():
if item.is_file() and item.stat().st_size <= _COPY_MAX_BYTES:
shutil.copy2(item, dest / rel / item.name)
for item in SOURCE.iterdir():
if item.is_file():
shutil.copy2(item, dest / item.name)
@pytest.fixture(scope="module")
def api(tmp_path_factory):
aiomysql = pytest.importorskip("aiomysql")
from fastapi import FastAPI
from fastapi.testclient import TestClient
import B05_Profile.B05_Profile_Router_Replan as replan
import common_util.common_util_storage as storage
import config.config_db as config_db
from config import config_system as cfg
storage_root = tmp_path_factory.mktemp("initial") / "storage"
project_root = storage_root / "1" / "3" / PROJECT_ID
_copy_project(project_root)
patch = pytest.MonkeyPatch()
patch.setattr(storage, "STORAGE_BASE_DIR", str(storage_root))
recomputed: list[str] = []
async def fake_recompute(project_id, root, stored_path):
recomputed.append(str(root))
return {"route_id": None, "total_length_m": None, "vertex_count": 0}
patch.setattr(replan, "_recompute", fake_recompute)
@asynccontextmanager
async def lifespan(_app):
try:
pool = await aiomysql.create_pool(
host=cfg.DB_HOST,
port=cfg.DB_PORT,
db=cfg.DB_NAME,
user=cfg.DB_USER,
password=cfg.DB_PASSWORD,
minsize=1,
maxsize=4,
autocommit=True,
charset="utf8mb4",
# 읽기 전용 세션 — 새는 쓰기가 있으면 DB 를 바꾸지 않고 터짐
init_command="SET time_zone='+00:00', SESSION tx_read_only=1",
)
except Exception as exc: # DB 가 안 닿는 곳
pytest.skip(f"DB 에 안 닿음: {exc}")
config_db.db_pool = pool
try:
yield
finally:
config_db.db_pool = None
pool.close()
await pool.wait_closed()
app = FastAPI(lifespan=lifespan)
app.include_router(replan.router)
try:
with TestClient(app) as client:
client.root = project_root # type: ignore[attr-defined]
client.recomputed = recomputed # type: ignore[attr-defined]
yield client
finally:
patch.undo()
def _screen(jobs: list[dict[str, Any]]) -> list[Any]:
if shutil.which("node") is None:
pytest.skip("node 가 없음")
out = subprocess.run(
["node", str(HELPER)],
input=json.dumps({"jobs": jobs}),
cwd=ROOT,
capture_output=True,
text=True,
encoding="utf-8",
timeout=120,
check=True,
)
return json.loads(out.stdout)
def _initial(api, logic: str, weights: dict[str, float] | None) -> dict[str, Any]:
body: dict[str, Any] = {"logic": logic}
if weights:
body["weights"] = weights
response = api.post(f"/api/projects/{PROJECT_ID}/route/initial", json=body)
assert response.status_code == 200, response.text
return response.json()
def _close(a: list[list[float]], b: list[list[float]], tol: float = 1e-3) -> bool:
return len(a) == len(b) and all(
abs(p[0] - q[0]) <= tol and abs(p[1] - q[1]) <= tol for p, q in zip(a, b)
)
@pytest.fixture(scope="module")
def rounds(api):
"""갈래마다 ①~③ — 결과를 모아 두고 시험마다 나눠 본다."""
out = []
before = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json()
for logic, weights in LOGICS:
answer = _initial(api, logic, weights)
[applied] = _screen(
[{"kind": "apply", "response": answer, "logic": logic, "weights": weights}]
)
confirmed = api.post(f"/api/projects/{PROJECT_ID}/route/replan", json=applied["body"])
assert confirmed.status_code == 200, confirmed.text
plan = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json()
[reopened] = _screen([{"kind": "reopen", "plan": plan, "cached": applied["choice"]}])
out.append(
{
"logic": logic,
"weights": weights,
"answer": answer,
"applied": applied,
"confirmed": confirmed.json(),
"plan": plan,
"reopened": reopened,
}
)
return {"before": before, "rounds": out}
def test_원본은_처음_추종(rounds):
assert rounds["before"]["current_initial"]["logic"] == "follow"
def test_비교값이_응답에_다_있음(rounds):
for item in rounds["rounds"]:
metrics = item["answer"]["metrics"]
for key in ("length_m", "max_grade_pct", "avg_grade_pct", "cut_m3", "fill_m3"):
assert isinstance(metrics[key], (int, float)), (item["logic"], key)
assert isinstance(metrics["over_limit"]["total"], int)
assert isinstance(metrics["offset_avg_m"], (int, float))
# 화면 한 줄에 빈 칸(—)이 없다.
assert "—" not in item["applied"]["metrics"], item["applied"]["metrics"]
def test_확인하면_서버가_초기값을_적고_다시_열면_그대로(rounds, api):
assert len(api.recomputed) == len(LOGICS)
for item in rounds["rounds"]:
sent = item["applied"]["body"]["initial"]
assert sent["mode"] == item["logic"]
assert item["confirmed"]["current_initial"]["logic"] == item["logic"]
current = item["plan"]["current_initial"]
assert current["logic"] == item["logic"]
if item["weights"]:
assert {k: round(v) for k, v in current["weights"].items()} == item["weights"]
reopened = item["reopened"]
assert reopened["base"]["logic"] == item["logic"]
assert reopened["label"] == item["applied"]["label"]
assert reopened["sameAsCache"] is True # [초기화]가 캐시 노드를 씀
saved = json.loads(
(api.root / "B05_Profile" / "route" / "initial_choice.json").read_text("utf-8")
)
assert saved["logic"] == "weighted"
def test_다시_연_노선은_확인한_노선(rounds):
for item in rounds["rounds"]:
sent = [[v["x"], v["y"]] for v in item["applied"]["body"]["vertices"]]
assert _close(item["reopened"]["planned"], sent), item["logic"]
def test_초기화는_마지막_초기값으로(rounds, api):
last = rounds["rounds"][-1]
cached = last["applied"]["choice"]["snapshot"]["planned"]
# 캐시가 있으면 그 노드 — 마지막 적용(비중 50/25/25).
assert _close(cached, [[v["x"], v["y"]] for v in last["applied"]["body"]["vertices"]])
# 캐시가 없으면(다른 PC) 서버가 적어 둔 갈래 · 비중으로 다시 물음 → 같은 노드.
base = last["reopened"]["base"]
again = _initial(api, base["logic"], base["weights"])
[redo] = _screen(
[{"kind": "apply", "response": again, "logic": base["logic"], "weights": base["weights"]}]
)
assert _close(redo["choice"]["snapshot"]["planned"], cached)
# 다른 갈래들과는 다른 노선이다(되돌림이 실제로 갈래를 가림).
others = [r["applied"]["choice"]["snapshot"]["planned"] for r in rounds["rounds"][:-1]]
assert all(not _close(o, cached) for o in others)
@@ -52,9 +52,9 @@ def test_계획고_편집은_가까운_새_측점으로_옮김(results):
assert "200.000" not in results["deleted"]["station_offsets"]
# ── 서버 — 사용자가 놓은 구조물 측점은 소수 그대로(격자 붙이기는 횡단 측점에만) ─────────
# 브레인 판정(다): 횡단 측점 · 파일 이름의 정수 m 격자 붙이기는 그대로 두고, 구조물 정본 측점만
# 사용자가 놓은 소수 그대로 저장 · 다시 읽기. 임시 storage 에서만 씀(공유 storage 안 건드림).
# ── 서버 — 관 · 구조물 자체 측점은 소수 그대로(정수 m 는 횡단 파일 이름에만) ─────────
# 브레인 판정(다) · 19장 남은 차이: 파일 이름은 한 정수 m 에 한 장 그대로, 측점 값은 사용자가 놓은
# 소수 그대로 저장 · 다시 읽기 · 종단 측점. 임시 storage 에서만 씀(공유 storage 안 건드림).
def _flat_sampler():
@@ -96,8 +96,8 @@ def test_사용자_구조물_119_8_저장_다시_읽기_그대로(tmp_path):
SectionGenerationOptions(extra_stations=extras),
)
chainages = [round(s["chainage_m"], 3) for s in sections["longitudinal"]["stations"]]
# 횡단 측점은 격자 120 으로 붙음(파일 이름 cross_00120m 한 장) — 정본 측점은 119.8 그대로
assert 120.0 in chainages and 119.8 not in chainages
# 횡단 측점도 자체 측점 119.8(같은 정수 미터 격자 120 대신 · 파일 이름 cross_00120m 한 장)
assert 119.8 in chainages and 120.0 not in chainages
assert cross_filename(119.8) == cross_filename(120.0)
_, again = load_structures(str(tmp_path))
assert [s.chainage_m for s in again] == [119.8] and again[0].memo == "메모만"