feat(D 예정노선): 옛 DB 설계 칸 읽기 걷음 · 지움 SQL 025(만들기만) — 설계 내용 = 폴더 파일만 (PLAN 590-3 · 요구 458 · 459)
- 옮김 끝 — 파일 없는 PLAN 10 개 = 옛 칸 → plan_network.json 한 번(일회 스크립트 · 11 개 모두 파일 대 DB 대조 같음) - 저장 모듈 = 파일 · 없으면 빈 설계(옛 칸 · 표 안 읽음) · 옛 연도 노선 접기(legacy_network) · 응답 legacy 칸 걷음 - D04 조각 노선 열쇠 route_json → route(옛 칸 이름과 헷갈림 걷음 · 메모리 값) - 복제 도구 = 옛 D 표 안 뜸 · 예정노선 설계는 폴더 파일로 따라감 - 024_plan_site_fill.py 지움 — 옛 칸을 읽는 일회 도구 · 기준점 11 개 모두 채움 - SQL 025 = 단계 번호(PLAN_BASE 걷음) · plan_networks · plan_route_pieces · projects.plan_base_json · 연도 줄 노선 칸 · 업체 줄 year 지움 — 공유 DB 적용 안 함(데스크탑 임시 배포가 새 코드를 받은 뒤 · 임시 DB 두 번 돌림 확인) - 시험: 590_1 옮김 시험 셋 → 파일 없음 = 빈 설계 · 코드에 옛 칸 이름 0 · 530_2 옛 연도 노선 시험 지움(그 길 걷음) · 591_2 옛 지점 꼴 시험 지움 · 591_1 채움 도구 시험 지움(도구 지움) · 573_1 = 파일째 따라감 · 옛 표 목록 밖 허용(025 전) · 531_3 · 563_2 · 593 = 열쇠 이름 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -64,8 +64,6 @@ export interface PlanNetworkView {
|
||||
lines: Array<{ id: string; length_m: number; stations: number[] }>;
|
||||
pieces: Array<PlanPiece & { length_m: number; planned: Vertex[] }>;
|
||||
} | null;
|
||||
/** 옛 연도 노선을 선으로 바꾼 것인가. */
|
||||
legacy: boolean;
|
||||
/** 사업연도 시작(있으면). */
|
||||
year_start?: number | null;
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
(요구 389 6 · 11 · 13 · 14 · 391 · 452 · PLAN 562-2 · 590-1).
|
||||
|
||||
- `GET /api/projects/{id}/plan/network` — 설계 내용 파일(`D03_PlanAnalysis_Store`) + 서버가 다시 센
|
||||
길이 · 측점 · 조각 · 파일이 없으면 옛 DB 칸을 옮겨 보임(쓰지 않음)
|
||||
길이 · 측점 · 조각 · 파일이 없으면 빈 설계
|
||||
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
|
||||
다시 셈 → 파일(원자 쓰기) · 연도 상태 줄(`plan_route_years` 지문 · 확정) · 사업연도 끝
|
||||
|
||||
@@ -213,51 +213,6 @@ def _sha1(value: Any) -> str:
|
||||
return hashlib.sha1(json.dumps(value, sort_keys=True).encode()).hexdigest() # noqa: S324
|
||||
|
||||
|
||||
def _json(value: Any) -> Any:
|
||||
if isinstance(value, (str, bytes)):
|
||||
try:
|
||||
return json.loads(value)
|
||||
except ValueError:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def legacy_network(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||||
"""옛 연도 노선(연도 차례 · 직전 종점 = 다음 시점) → 본선 하나 + 연도 경계 마크 + 연도 조각."""
|
||||
vertices: list[list[float]] = []
|
||||
pieces: list[dict[str, Any]] = []
|
||||
epsg = None
|
||||
for row in rows:
|
||||
route = _json(row.get("route_json"))
|
||||
planned = route.get("planned") if isinstance(route, dict) else None
|
||||
if not planned or len(planned) < 2:
|
||||
continue
|
||||
epsg = epsg or row.get("epsg")
|
||||
points = [[float(p[0]), float(p[1])] for p in planned]
|
||||
if vertices and draw.segment_length(vertices[-1], points[0]) <= SAME_M:
|
||||
points = points[1:]
|
||||
start = draw.line_length(vertices)
|
||||
vertices.extend(points)
|
||||
end = draw.line_length(vertices)
|
||||
pieces.append(
|
||||
{
|
||||
"id": f"y{row['year']}",
|
||||
"line": "main",
|
||||
"from_m": start,
|
||||
"to_m": end,
|
||||
"year": row["year"],
|
||||
}
|
||||
)
|
||||
if not pieces:
|
||||
return None
|
||||
return {
|
||||
"epsg": int(epsg) if epsg is not None else None,
|
||||
"lines": [{"id": "main", "vertices": vertices, "branch": None}],
|
||||
"marks": [{"line": "main", "at_m": p["to_m"]} for p in pieces[:-1]],
|
||||
"pieces": pieces,
|
||||
}
|
||||
|
||||
|
||||
def work_epsg(plan: dict[str, Any], site: dict[str, Any] | None) -> int | None:
|
||||
"""작업 좌표계 — 파일 머리 · 없으면 기준점 경도의 TM 띠(첫 [저장] · 지형 받기 때 굳음)."""
|
||||
if plan.get("epsg") is not None:
|
||||
@@ -285,7 +240,6 @@ def _view(plan: dict[str, Any], project: dict[str, Any]) -> dict[str, Any]:
|
||||
"year_start": project["year_start"],
|
||||
"network": network if plan["lines"] else None,
|
||||
"derived": derive(network) if plan["lines"] else None,
|
||||
"legacy": False,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -6,8 +6,7 @@
|
||||
연도 지문은 읽을 때 `plan_state` 가 셈(저장 안 함).
|
||||
- DB 에 남는 것 = 등록 칸 · 연도 상태(`plan_route_years` 의 year · fingerprint · confirmed_at) ·
|
||||
업체 줄(`plan_route_assignees` 조각 번호 · 업체 · 메일 시각 · 연결 — 연도는 이 파일에서).
|
||||
- 옮김 = 파일이 없으면 옛 DB 칸(`projects.plan_base_json` · `plan_networks` · `plan_route_years`)을
|
||||
메모리에서 새 꼴로(`migrate`) · 파일은 첫 쓰기 때 생김 · 옛 칸은 손대지 않음(590-3 에서 지움).
|
||||
- 파일이 없으면 빈 설계(새 프로젝트) · 파일은 첫 쓰기 때 생김(옛 DB 칸 옮김 = 590-3 에서 끝남).
|
||||
- 쓰기 = `project_lock(project_id)` 안에서 `save_plan`(같은 폴더 임시 파일 → 바꿔치기 · 반쯤 쓴
|
||||
파일이 안 보임) · 읽기는 잠금 없음.
|
||||
- 확정 = 연도 상태 줄 지문이 파일 지문과 같을 때만(파일만 바뀌고 DB 가 못 따라간 틈 =
|
||||
@@ -24,7 +23,6 @@ from uuid import UUID
|
||||
import aiomysql
|
||||
from fastapi import HTTPException
|
||||
|
||||
from common_util.common_util_crs import lonlat_to_tm, tm_to_lonlat
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from config.config_db import get_db_pool
|
||||
@@ -66,74 +64,9 @@ def save_plan(root: str | Path, plan: dict[str, Any]) -> None:
|
||||
atomic_write_json(plan_path(root), _clean(plan))
|
||||
|
||||
|
||||
def _parsed(value: Any) -> Any:
|
||||
if isinstance(value, (str, bytes)):
|
||||
try:
|
||||
return json.loads(value)
|
||||
except ValueError:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _old_points(base: dict[str, Any] | None, rows: list[dict[str, Any]]) -> tuple[Any, list]:
|
||||
"""옛 지점 칸 → (epsg, [{x, y}]) — points · 옛 starts/ends 목록 · 더 옛 start/end 한 점 ·
|
||||
칸이 비면 [첫 연도 시점, 마지막 연도 종점]으로 접음."""
|
||||
if base:
|
||||
items = base.get("points")
|
||||
if items is None:
|
||||
starts = base.get("starts", [base["start"]] if base.get("start") else [])
|
||||
ends = base.get("ends", [base["end"]] if base.get("end") else [])
|
||||
items = [*starts, *ends]
|
||||
epsg = base.get("epsg")
|
||||
else:
|
||||
dated = [r for r in rows if r.get("start_x") is not None or r.get("end_x") is not None]
|
||||
if not dated:
|
||||
return None, []
|
||||
first, last = dated[0], dated[-1]
|
||||
epsg = first.get("epsg")
|
||||
items = [
|
||||
{"x": first.get("start_x"), "y": first.get("start_y")},
|
||||
{"x": last.get("end_x"), "y": last.get("end_y")},
|
||||
]
|
||||
points = [
|
||||
{"x": float(p["x"]), "y": float(p["y"])}
|
||||
for p in items
|
||||
if p and p.get("x") is not None and p.get("y") is not None
|
||||
]
|
||||
return (epsg if points else None), points
|
||||
|
||||
|
||||
def migrate(base_json: Any, network_json: Any, year_rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""옛 DB 칸 → 새 꼴(메모리) — 노선 = `plan_networks` · 없으면 옛 연도 노선을 한 선으로."""
|
||||
rows = sorted(year_rows, key=lambda r: int(r["year"]))
|
||||
epsg, points = _old_points(_parsed(base_json), rows)
|
||||
net = _parsed(network_json) or network.legacy_network(rows)
|
||||
plan = empty_plan(int(epsg) if epsg is not None else None)
|
||||
plan["points"] = points
|
||||
if net and net.get("lines"):
|
||||
net_epsg = int(net["epsg"]) if net.get("epsg") is not None else plan["epsg"]
|
||||
if points and plan["epsg"] is not None and net_epsg != plan["epsg"]:
|
||||
plan["points"] = [
|
||||
dict(
|
||||
zip(
|
||||
("x", "y"),
|
||||
lonlat_to_tm(*tm_to_lonlat(p["x"], p["y"], plan["epsg"]), net_epsg),
|
||||
)
|
||||
)
|
||||
for p in points
|
||||
]
|
||||
plan.update(
|
||||
epsg=net_epsg,
|
||||
lines=net.get("lines", []),
|
||||
marks=net.get("marks", []),
|
||||
pieces=net.get("pieces", []),
|
||||
)
|
||||
return plan
|
||||
|
||||
|
||||
async def _project(cursor: Any, project_id: UUID | str) -> dict[str, Any]:
|
||||
await cursor.execute(
|
||||
"SELECT project_kind, project_year, storage_path, plan_base_json FROM projects"
|
||||
"SELECT project_kind, project_year, storage_path FROM projects"
|
||||
" WHERE id = %s AND deleted_at IS NULL",
|
||||
(str(project_id),),
|
||||
)
|
||||
@@ -148,26 +81,14 @@ async def _project(cursor: Any, project_id: UUID | str) -> dict[str, Any]:
|
||||
|
||||
|
||||
async def load_plan(project_id: UUID | str) -> tuple[Path, dict[str, Any]]:
|
||||
"""(프로젝트 실경로, 설계 내용) — 파일 · 없으면 옛 DB 칸을 옮긴 것(쓰지 않음)."""
|
||||
pid = str(project_id)
|
||||
"""(프로젝트 실경로, 설계 내용) — 파일 · 없으면 빈 설계(쓰지 않음)."""
|
||||
async with (
|
||||
get_db_pool().acquire() as connection,
|
||||
connection.cursor(aiomysql.DictCursor) as cursor,
|
||||
):
|
||||
project = await _project(cursor, pid)
|
||||
root = Path(resolve_stored_project_path(project["storage_path"]))
|
||||
plan = read_plan(root)
|
||||
if plan is not None:
|
||||
return root, plan
|
||||
await cursor.execute("SELECT network_json FROM plan_networks WHERE project_id = %s", (pid,))
|
||||
row = await cursor.fetchone()
|
||||
await cursor.execute(
|
||||
"SELECT year, start_x, start_y, end_x, end_y, epsg, route_json FROM plan_route_years"
|
||||
" WHERE project_id = %s ORDER BY year",
|
||||
(pid,),
|
||||
)
|
||||
years = list(await cursor.fetchall())
|
||||
return root, migrate(project["plan_base_json"], row and row["network_json"], years)
|
||||
project = await _project(cursor, project_id)
|
||||
root = Path(resolve_stored_project_path(project["storage_path"]))
|
||||
return root, read_plan(root) or empty_plan()
|
||||
|
||||
|
||||
def plan_units(plan: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
|
||||
@@ -68,14 +68,14 @@ class LinkRequest(BaseModel):
|
||||
las_free: bool = False # B03 「LAS 없이 설계」(화면 기억값) — 입력 완료 판정에만
|
||||
|
||||
|
||||
def _parsed(route_json: Any) -> dict[str, Any] | None:
|
||||
value = json.loads(route_json) if isinstance(route_json, str | bytes) else route_json
|
||||
def _parsed(route: Any) -> dict[str, Any] | None:
|
||||
value = json.loads(route) if isinstance(route, str | bytes) else route
|
||||
return value if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def route_fingerprint(route_json: Any) -> str | None:
|
||||
def route_fingerprint(route: Any) -> str | None:
|
||||
"""연도 노선 내용 지문(sha1) — 고른 중심선 점(mm 반올림)만 봄 · 같은 값이면 같은 지문."""
|
||||
body = _parsed(route_json)
|
||||
body = _parsed(route)
|
||||
planned = (body or {}).get("planned")
|
||||
if not planned:
|
||||
return None
|
||||
@@ -83,14 +83,12 @@ def route_fingerprint(route_json: Any) -> str | None:
|
||||
return hashlib.sha1(json.dumps(points).encode()).hexdigest()
|
||||
|
||||
|
||||
def update_available(
|
||||
route_json: Any, source_fingerprint: str | None, confirmed: bool = True
|
||||
) -> bool:
|
||||
def update_available(route: Any, source_fingerprint: str | None, confirmed: bool = True) -> bool:
|
||||
"""연결 때 지문 ≠ 지금 지문 — 지금 노선이 비면(지점 바뀜 · 분석 전) 알리지 않음.
|
||||
|
||||
확정이 풀린 연도(D02 지점 바뀜 뒤 다시 [저장]만)도 알리지 않음 — 가져오기가 409 라서.
|
||||
"""
|
||||
now = route_fingerprint(route_json)
|
||||
now = route_fingerprint(route)
|
||||
return confirmed and bool(source_fingerprint) and now is not None and now != source_fingerprint
|
||||
|
||||
|
||||
@@ -104,7 +102,7 @@ async def _fetch_one(sql: str, args: tuple) -> dict[str, Any] | None:
|
||||
|
||||
|
||||
async def piece_route(project_id: str, piece_id: str | None) -> dict[str, Any]:
|
||||
"""그 조각(파일) — {year, epsg, route_json: {planned, 누가 · 길이}, confirmed_at} ·
|
||||
"""그 조각(파일) — {year, epsg, route: {planned, 누가 · 길이}, confirmed_at} ·
|
||||
조각이 없으면 노선 · 연도 · 확정 None."""
|
||||
state = await store.plan_state(project_id)
|
||||
unit = next((u for u in state["units"] if u["id"] == piece_id), None)
|
||||
@@ -112,13 +110,13 @@ async def piece_route(project_id: str, piece_id: str | None) -> dict[str, Any]:
|
||||
return {
|
||||
"year": None,
|
||||
"epsg": state["plan"]["epsg"],
|
||||
"route_json": None,
|
||||
"route": None,
|
||||
"confirmed_at": None,
|
||||
}
|
||||
return {
|
||||
"year": unit["year"],
|
||||
"epsg": unit["epsg"],
|
||||
"route_json": {
|
||||
"route": {
|
||||
"planned": unit["planned"],
|
||||
"start_chainage_m": unit["from_m"],
|
||||
"end_chainage_m": unit["to_m"],
|
||||
@@ -175,7 +173,7 @@ async def record_link(project_id: str, assignee: dict[str, Any], fingerprint: st
|
||||
|
||||
def year_files(assignee: dict[str, Any]) -> dict[str, bytes]:
|
||||
"""그 조각 shp 한 벌(.shp .shx .dbf .prj .cpg) — 파일명 → 바이트."""
|
||||
body = _parsed(assignee["route_json"]) or {}
|
||||
body = _parsed(assignee["route"]) or {}
|
||||
route = YearRoute(
|
||||
year=int(assignee["year"]),
|
||||
points=[(p[0], p[1]) for p in body["planned"]],
|
||||
@@ -278,7 +276,7 @@ async def import_route_files(
|
||||
|
||||
async def _new_road_project(assignee: dict[str, Any], session: dict[str, Any]) -> str:
|
||||
"""예정노선 이름 · 지역 · 임도 종류 · 그 연도로 임도 프로젝트(B02 등록과 같은 함수)."""
|
||||
body = _parsed(assignee["route_json"]) or {}
|
||||
body = _parsed(assignee["route"]) or {}
|
||||
# 「2026 홍천 노선 계획」 꼴 = 앞 연도를 그 연도로 갈아 끼움(연도 두 번 안 붙음) · 그 밖 = 뒤에 연도
|
||||
name, dated = re.subn(r"^\d{4}(?:~\d{4})?(?=\s)", str(assignee["year"]), assignee["name"])
|
||||
created = await create_project(
|
||||
@@ -304,7 +302,7 @@ async def link_plan_route(
|
||||
raise HTTPException(status_code=404, detail="지정 업체 행을 찾을 수 없습니다.")
|
||||
if assignee["company_id"] is None or assignee["company_id"] != session.get("company_id"):
|
||||
raise HTTPException(status_code=403, detail="이 예정노선을 지정받은 회사가 아닙니다.")
|
||||
if not assignee["confirmed_at"] or route_fingerprint(assignee["route_json"]) is None:
|
||||
if not assignee["confirmed_at"] or route_fingerprint(assignee["route"]) is None:
|
||||
raise HTTPException(status_code=409, detail="확정된 노선이 없습니다.")
|
||||
if payload.mode == "existing":
|
||||
if payload.project_id is None:
|
||||
@@ -329,7 +327,7 @@ async def link_plan_route(
|
||||
return _confirm_replace_response(exc)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
fingerprint = route_fingerprint(assignee["route_json"])
|
||||
fingerprint = route_fingerprint(assignee["route"])
|
||||
await record_link(project_id, assignee, fingerprint)
|
||||
return {
|
||||
"status": "success",
|
||||
@@ -364,6 +362,6 @@ async def plan_source(project_id: UUID) -> dict[str, Any]:
|
||||
"year": row["year"],
|
||||
"piece_id": row.get("piece_id"),
|
||||
"update_available": update_available(
|
||||
row["route_json"], row["source_fingerprint"], row["confirmed_at"] is not None
|
||||
row["route"], row["source_fingerprint"], row["confirmed_at"] is not None
|
||||
),
|
||||
}
|
||||
|
||||
@@ -186,14 +186,11 @@ def visible_to(
|
||||
|
||||
|
||||
async def _units(project_id: UUID) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||
"""(연도 줄, 단위) — 파일에서 · 연도 줄 = {year, epsg, route_json, fingerprint,
|
||||
"""(연도 줄, 단위) — 파일에서 · 연도 줄 = {year, epsg, route, fingerprint,
|
||||
confirmed_at}."""
|
||||
state = await store.plan_state(project_id)
|
||||
epsg = state["plan"]["epsg"]
|
||||
rows = [
|
||||
{"year": year, "epsg": epsg, "route_json": row["route"], **row}
|
||||
for year, row in state["years"].items()
|
||||
]
|
||||
rows = [{"year": year, "epsg": epsg, **row} for year, row in state["years"].items()]
|
||||
return rows, route_units(state["units"])
|
||||
|
||||
|
||||
@@ -334,7 +331,7 @@ def _end_point(point: list[float], epsg: int) -> dict[str, float]:
|
||||
|
||||
def year_summary(row: dict[str, Any]) -> dict[str, Any]:
|
||||
"""연도 한 줄 — 분석 [저장]이 없으면 saved False 만."""
|
||||
body = row.get("route_json") or {}
|
||||
body = row.get("route") or {}
|
||||
planned = body.get("planned") or []
|
||||
out = {"year": int(row["year"]), "saved": len(planned) >= 2, "confirmed_at": None}
|
||||
if row.get("confirmed_at"):
|
||||
|
||||
@@ -4,9 +4,9 @@
|
||||
|
||||
full DB 행(projects · project_members · input_files · processed_point_cloud ·
|
||||
surface_models · terrain_layers · routes · route_points · route_statistics ·
|
||||
longitudinal_sections · cross_sections · project_workflow_stages · D 의 plan_networks ·
|
||||
plan_route_pieces · plan_route_years · plan_route_assignees)을 새 id 로 다시 이어
|
||||
한 트랜잭션에 넣고 저장소 폴더 통째(robocopy) — 복제 도구(`db_management/
|
||||
longitudinal_sections · cross_sections · project_workflow_stages · D 의
|
||||
plan_route_years · plan_route_assignees)을 새 id 로 다시 이어 한 트랜잭션에 넣고
|
||||
저장소 폴더 통째(robocopy · D 설계 내용 = 폴더 파일) — 복제 도구(`db_management/
|
||||
tools_clone_project.py`)의 본체. 종류(B · D · E)는 `project_kind` 로 판별 — 같은 길로
|
||||
뜨고(빈 표는 건너뜀) E 의 원본 칸(source_project_id 등)은 projects 행째 그대로.
|
||||
stability DB 행은 넣지 않고 E 프로젝트 저장소 `E01_StabRegister/source/` 에 동결 스냅샷
|
||||
@@ -36,7 +36,7 @@ from common_util.common_util_time import utc_now_text
|
||||
Scope = Literal["full", "stability"]
|
||||
KIND_LABELS = {"ROAD": "B", "PLAN": "D", "STABILITY": "E"}
|
||||
#: 자동증가 id 없이 project_id 만 바꿔 넣는 D 표
|
||||
_PLAN_TABLES = ("plan_networks", "plan_route_pieces", "plan_route_years")
|
||||
_PLAN_TABLES = ("plan_route_years",)
|
||||
_PLAN_ALL = (*_PLAN_TABLES, "plan_route_assignees")
|
||||
MANIFEST_NAME = "copy_manifest.json"
|
||||
#: E 프로젝트 안 복사본 자리 — E 엔진은 `E01_StabRegister_Source` 로만 읽음
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
-- 024_plan_site.sql
|
||||
-- 예정노선 등록 기준점 하나 + 주소 · 설계속도 비움 (요구 452 · 453 · PLAN 591-1)
|
||||
--
|
||||
-- 추가만 함(칸 셋) · 옛 PLAN 기준점 = 지점 가운데 → 이 파일 뒤 `024_plan_site_fill.py` 한 번
|
||||
-- 추가만 함(칸 셋) · 옛 PLAN 기준점 = 지점 가운데 → 이 파일 뒤 채움 도구 한 번(590-3 에서 걷음 · 모두 채움)
|
||||
-- 설계속도 칸은 남김(옛 코드 대시보드가 SELECT) · 예정노선 값만 비움 = B 로 넘길 때 B05 기본 20
|
||||
|
||||
USE aislo_db;
|
||||
|
||||
@@ -1,74 +0,0 @@
|
||||
"""옛 예정노선 기준점 채움 — 지점 상자 가운데 · 주소 VWorld (PLAN 591-1 · SQL 024 뒤 한 번).
|
||||
|
||||
기준점 칸이 빈 PLAN 만 · 지점 = `plan_base_json`(새 꼴 points · 옛 start/end · starts/ends) ·
|
||||
지점이 없으면 건너뜀(알림 한 줄) · 다시 돌려도 같음(채운 행은 안 건드림).
|
||||
|
||||
쓰는 법
|
||||
./venv/Scripts/python.exe db_management/024_plan_site_fill.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from dotenv import load_dotenv # noqa: E402
|
||||
|
||||
load_dotenv()
|
||||
|
||||
import aiomysql # noqa: E402
|
||||
|
||||
from B01_Dashboard.B01_Dashboard_Map import reverse_place # noqa: E402
|
||||
from common_util.common_util_crs import tm_to_lonlat # noqa: E402
|
||||
from config.config_db import close_db_pool, init_db_pool # noqa: E402
|
||||
|
||||
|
||||
def base_center(base: dict[str, Any] | None) -> tuple[float, float] | None:
|
||||
"""지점 상자 가운데(TM m) — 꼴 셋을 모두 읽음 · 지점 없음 = None."""
|
||||
base = base or {}
|
||||
points = [
|
||||
*base.get("points", []),
|
||||
*base.get("starts", []),
|
||||
*base.get("ends", []),
|
||||
*(base[k] for k in ("start", "end") if base.get(k)),
|
||||
]
|
||||
xs = [float(p["x"]) for p in points if p]
|
||||
ys = [float(p["y"]) for p in points if p]
|
||||
if not xs:
|
||||
return None
|
||||
return (min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
pool = await init_db_pool()
|
||||
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT id, name, plan_base_json FROM projects"
|
||||
" WHERE project_kind = 'PLAN' AND site_lon IS NULL"
|
||||
)
|
||||
for row in await cursor.fetchall():
|
||||
raw = row["plan_base_json"]
|
||||
base = json.loads(raw) if isinstance(raw, str) else raw
|
||||
center = base_center(base)
|
||||
if not center or not (base or {}).get("epsg"):
|
||||
print(f"건너뜀(지점 없음) {row['id']} {row['name']}")
|
||||
continue
|
||||
lon, lat = tm_to_lonlat(*center, int(base["epsg"]))
|
||||
address = (await reverse_place(lon, lat))["address"]
|
||||
await cursor.execute(
|
||||
"UPDATE projects SET site_lon = %s, site_lat = %s, site_address = %s"
|
||||
" WHERE id = %s AND site_lon IS NULL",
|
||||
(lon, lat, address, row["id"]),
|
||||
)
|
||||
print(f"채움 {row['name']} {lon:.6f} {lat:.6f} {address}")
|
||||
await connection.commit()
|
||||
await close_db_pool()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,43 @@
|
||||
-- 025_plan_old_columns.sql
|
||||
-- 예정노선 옛 설계 칸 · 표 지움 (요구 458 · 459 · PLAN 590-3)
|
||||
--
|
||||
-- 설계 내용 = 프로젝트 폴더 `D03_PlanAnalysis/plan_network.json`(023 · 590-3 에서 11 개 모두 옮김 · 대조 같음)
|
||||
-- 적용 = 데스크탑 임시 배포가 590-3 코드를 받은 뒤(옛 코드가 이 칸을 읽음) · 다시 돌려도 같음
|
||||
-- 남김 = plan_route_years(year · fingerprint · confirmed_at) · plan_route_assignees(조각 번호 · 업체 · 메일 · 연결)
|
||||
|
||||
USE aislo_db;
|
||||
|
||||
-- 1) 단계 번호(PLAN_BASE 걷음 · 591-2) — PLAN_ANALYSIS 0 · PLAN_CONFIRM 1 · 상태 그대로
|
||||
DELETE w FROM project_workflow_stages w
|
||||
JOIN projects p ON p.id = w.project_id
|
||||
WHERE p.project_kind = 'PLAN' AND w.stage_key = 'PLAN_BASE';
|
||||
|
||||
UPDATE project_workflow_stages w
|
||||
JOIN projects p ON p.id = w.project_id
|
||||
SET w.stage_no = w.stage_no + 100
|
||||
WHERE p.project_kind = 'PLAN' AND w.stage_no < 100;
|
||||
|
||||
UPDATE project_workflow_stages w
|
||||
JOIN projects p ON p.id = w.project_id
|
||||
SET w.stage_no = CASE w.stage_key WHEN 'PLAN_ANALYSIS' THEN 0 WHEN 'PLAN_CONFIRM' THEN 1 END
|
||||
WHERE p.project_kind = 'PLAN' AND w.stage_no >= 100;
|
||||
|
||||
-- 2) 옛 설계 표 · 칸
|
||||
DROP TABLE IF EXISTS plan_route_pieces;
|
||||
DROP TABLE IF EXISTS plan_networks;
|
||||
|
||||
ALTER TABLE projects DROP COLUMN IF EXISTS plan_base_json;
|
||||
|
||||
ALTER TABLE plan_route_years
|
||||
DROP COLUMN IF EXISTS start_x,
|
||||
DROP COLUMN IF EXISTS start_y,
|
||||
DROP COLUMN IF EXISTS end_x,
|
||||
DROP COLUMN IF EXISTS end_y,
|
||||
DROP COLUMN IF EXISTS epsg,
|
||||
DROP COLUMN IF EXISTS route_json,
|
||||
COMMENT = '예정노선 연도 상태(지문 · 확정)';
|
||||
|
||||
-- 업체 줄 연도 = 파일 조각 연도(023) · 연도 색인 = 조각 색인(project_id, piece_id)이 갈음
|
||||
ALTER TABLE plan_route_assignees
|
||||
DROP INDEX IF EXISTS idx_plan_route_assignees_project,
|
||||
DROP COLUMN IF EXISTS year;
|
||||
@@ -11,8 +11,8 @@
|
||||
무엇을 뜨나
|
||||
DB projects · project_members · input_files · processed_point_cloud · surface_models ·
|
||||
terrain_layers · routes · route_points · route_statistics · longitudinal_sections ·
|
||||
cross_sections · project_workflow_stages · (D) plan_networks · plan_route_pieces ·
|
||||
plan_route_years · plan_route_assignees — 종류(B · D · E)는 project_kind 로 판별
|
||||
cross_sections · project_workflow_stages · (D) plan_route_years · plan_route_assignees —
|
||||
종류(B · D · E)는 project_kind 로 판별 · D 설계 내용 = 폴더 파일(아래 파일째 따라감)
|
||||
⚠ 자동증가 id 는 **새로 받고** 참조(외래키)를 새 id 로 다시 이어 붙인다.
|
||||
파일 저장소 폴더 통째(robocopy). 경로가 프로젝트 루트 기준 상대라 그대로 쓰인다.
|
||||
치환 파일 안에 박힌 **옛 프로젝트 UUID**, 초기값 스냅숏의 **surface_model_id**.
|
||||
|
||||
@@ -106,23 +106,6 @@ def test_year_rows_group_pieces() -> None:
|
||||
assert rows[2027]["route"]["start_chainage_m"] == 75.0
|
||||
|
||||
|
||||
def test_legacy_rows_become_one_line() -> None:
|
||||
rows = [
|
||||
{"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]})},
|
||||
{"year": 2027, "epsg": 5186, "route_json": None},
|
||||
{"year": 2028, "epsg": 5186, "route_json": {"planned": [[30, 40], [30, 100]]}},
|
||||
]
|
||||
got = net.legacy_network(rows)
|
||||
assert got["lines"][0]["vertices"] == [[0, 0], [30, 40], [30, 100]]
|
||||
assert got["marks"] == [{"line": "main", "at_m": 50.0}]
|
||||
assert [(p["year"], p["from_m"], p["to_m"]) for p in got["pieces"]] == [
|
||||
(2026, 0, 50),
|
||||
(2028, 50, 110),
|
||||
]
|
||||
assert net.derive(got)["pieces"][1]["planned"] == [[30, 40], [30, 100]]
|
||||
assert net.legacy_network([{"year": 2026, "epsg": 5186, "route_json": None}]) is None
|
||||
|
||||
|
||||
class _Cursor:
|
||||
def __init__(self, db):
|
||||
self.db = db
|
||||
@@ -188,14 +171,14 @@ BASE = {"epsg": 5186, "points": [{"x": 0.0, "y": 0.0}, {"x": 90.0, "y": 180.0}]}
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch, tmp_path):
|
||||
"""파일 = tmp_path 진짜 파일 · 없으면 옛 DB 칸(`old`)을 옮김 · DB = 가짜 기록."""
|
||||
"""파일 = tmp_path 진짜 파일 · 없으면 `old` 설계(지점만) · DB = 가짜 기록."""
|
||||
db = _Db()
|
||||
monkeypatch.setattr(net, "get_db_pool", lambda: db)
|
||||
old = {"base": json.dumps(BASE), "network": None, "rows": []}
|
||||
old = {**store.empty_plan(5186), "points": BASE["points"]}
|
||||
|
||||
async def load(project_id):
|
||||
plan = store.read_plan(tmp_path)
|
||||
return tmp_path, plan or store.migrate(old["base"], old["network"], old["rows"])
|
||||
return tmp_path, plan or dict(old)
|
||||
|
||||
monkeypatch.setattr(store, "load_plan", load)
|
||||
app = FastAPI()
|
||||
@@ -213,14 +196,9 @@ def client(monkeypatch, tmp_path):
|
||||
URL = f"/api/projects/{PID}/plan/network"
|
||||
|
||||
|
||||
def test_get_migrates_old_db_or_empty(client) -> None:
|
||||
def test_get_reads_plan_or_empty(client) -> None:
|
||||
assert client.get(URL).json()["network"] is None
|
||||
client.old["rows"] = [
|
||||
{"year": 2026, "epsg": 5186, "route_json": '{"planned": [[0, 0], [30, 40]]}'}
|
||||
]
|
||||
got = client.get(URL).json()
|
||||
assert got["legacy"] is False and got["derived"]["pieces"][0]["length_m"] == 50
|
||||
client.old["network"] = json.dumps(_network())
|
||||
client.old.update(_network())
|
||||
got = client.get(URL).json()
|
||||
assert len(got["derived"]["pieces"]) == 3 and got["network"]["epsg"] == 5186
|
||||
assert got["base"]["points"][1]["x"] == 90.0 and got["year_start"] == 2026
|
||||
@@ -233,7 +211,7 @@ def test_put_writes_file_and_year_status_only(client) -> None:
|
||||
assert got.json()["year_start"] == 2026
|
||||
saved = json.loads(client.file.read_text("utf-8"))
|
||||
assert saved["version"] == 1 and saved["epsg"] == 5186
|
||||
assert saved["points"] == BASE["points"] # 옛 지점 칸 → 파일로 옮김
|
||||
assert saved["points"] == BASE["points"] # 몸에 지점 없음 = 그대로
|
||||
assert [p["id"] for p in saved["pieces"]] == ["p1", "p2", "p3"]
|
||||
assert "length_m" not in json.dumps(saved) # 파생 저장 안 함
|
||||
log = client.db.log
|
||||
|
||||
@@ -71,7 +71,7 @@ def _assignee(**kw):
|
||||
"road_type": "work",
|
||||
"route_start_m": 0.0,
|
||||
"epsg": 5186,
|
||||
"route_json": json.dumps(_route(P2026)),
|
||||
"route": json.dumps(_route(P2026)),
|
||||
"confirmed_at": "2026-10-25 12:00:00",
|
||||
}
|
||||
row.update(kw)
|
||||
@@ -212,7 +212,7 @@ def test_link_writes_route_set_like_upload(env, monkeypatch):
|
||||
assert meta["point_count"] == len(P2026)
|
||||
stage_meta = json.loads((env["root"] / "B03_FileInput/metadata.json").read_text("utf-8"))
|
||||
assert len(stage_meta["files"]) == 5
|
||||
fp = link.route_fingerprint(_assignee()["route_json"])
|
||||
fp = link.route_fingerprint(_assignee()["route"])
|
||||
assert env["links"] == [(ROAD_ID, 7, fp)] and len(fp) == 40
|
||||
assert any("SUPERSEDED" in sql for sql, _ in env["log"] if isinstance(sql, str))
|
||||
|
||||
@@ -325,7 +325,7 @@ def test_assigned_list_flags_update(plan_file, monkeypatch):
|
||||
(8, 2026, False),
|
||||
(9, 2027, False),
|
||||
]
|
||||
assert "route_json" not in out[0] and "source_fingerprint" not in out[0]
|
||||
assert "route" not in out[0] and "source_fingerprint" not in out[0]
|
||||
|
||||
|
||||
def test_plan_source(plan_file, monkeypatch):
|
||||
|
||||
@@ -218,7 +218,7 @@ def test_confirm_drops_when_file_changes(plan, monkeypatch):
|
||||
def test_link_uses_piece_route(plan):
|
||||
unit = asyncio.run(link.piece_route(PLAN_ID, "p3"))
|
||||
assert unit["year"] == 2026 and unit["epsg"] == 5186 and unit["confirmed_at"] == "x"
|
||||
assert unit["route_json"]["planned"] == P3 and unit["route_json"]["length_m"] == 130.0
|
||||
assert unit["route"]["planned"] == P3 and unit["route"]["length_m"] == 130.0
|
||||
files = link.year_files({**unit, "name": "2026 홍천 노선 계획", "region": "일월.용화"})
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
for name, data in files.items():
|
||||
@@ -226,7 +226,7 @@ def test_link_uses_piece_route(plan):
|
||||
meta = analyze_input_metadata(next(Path(tmp).glob("*.shp")))
|
||||
assert meta["point_count"] == 2 and float(meta["attributes"]["종점누가"]) == 130.0
|
||||
gone = asyncio.run(link.piece_route(PLAN_ID, "없음"))
|
||||
assert gone["route_json"] is None and gone["year"] is None
|
||||
assert gone["route"] is None and gone["year"] is None
|
||||
|
||||
|
||||
class _Cursor:
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""573-1 복제 도구 D · E 갈래 — 임시 storage 사본에서(요구 404 · 405).
|
||||
|
||||
- D(예정노선) — plan_networks · plan_route_pieces · plan_route_years · plan_route_assignees ·
|
||||
plan_base_json · 지표면 · 지형층 행 수 같음 · 자동증가 id 새로 · 지형층 → 새 지표면 ·
|
||||
- D(예정노선) — plan_route_years · plan_route_assignees · 지표면 · 지형층 행 수 같음 ·
|
||||
설계 내용 = 폴더 파일(plan_network.json) 그대로 따라감 · 자동증가 id 새로 · 지형층 → 새 지표면 ·
|
||||
업체 연결 끊김 · 지형 models_plan · _derived 파일 그대로
|
||||
- E(안정성평가) — 원본 칸(source_project_id 등) 그대로 · 단계 행 여섯 상태 그대로 ·
|
||||
E05 보고서 · _derived 파일 수 같음
|
||||
@@ -34,6 +34,8 @@ from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER #
|
||||
PLAN_ID = "87d47800-9936-4a56-9850-276fbcb34cac"
|
||||
STAB_ID = "088e173d-a8ab-4c51-badb-f5a95afc43bb"
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
#: 복제 안 뜨는 옛 D 표 — 지움 SQL 025 적용 전까지 남음(설계 내용 = 폴더 파일)
|
||||
RETIRED = {"plan_networks", "plan_route_pieces"}
|
||||
TABLES = (
|
||||
"surface_models",
|
||||
"project_members",
|
||||
@@ -159,7 +161,7 @@ def _common(source, root, new_id):
|
||||
new_root = root.parent / new_id
|
||||
manifest = json.loads((new_root / cp.MANIFEST_NAME).read_text(encoding="utf-8"))
|
||||
assert manifest["kind"] == cp.KIND_LABELS[source["project_kind"]]
|
||||
assert manifest["skipped_tables"] == {}
|
||||
assert set(manifest["skipped_tables"]) <= RETIRED
|
||||
assert _files(new_root) - {cp.MANIFEST_NAME} == _files(root)
|
||||
for path in new_root.rglob("*.json"):
|
||||
if path.name != cp.MANIFEST_NAME:
|
||||
@@ -170,7 +172,8 @@ def _common(source, root, new_id):
|
||||
def test_clone_plan(monkeypatch):
|
||||
with _clone(PLAN_ID, "시험_573_1_D", monkeypatch) as (source, root, new_id):
|
||||
new, new_root = _common(source, root, new_id)
|
||||
assert json.loads(new["plan_base_json"]) == json.loads(source["plan_base_json"])
|
||||
plan = Path("D03_PlanAnalysis") / "plan_network.json"
|
||||
assert (new_root / plan).read_bytes() == (root / plan).read_bytes()
|
||||
assert (new_root / "B04_PreProcess" / "models_plan").is_dir()
|
||||
assert (new_root / "_derived").is_dir()
|
||||
|
||||
@@ -179,22 +182,19 @@ def test_clone_plan(monkeypatch):
|
||||
f"SELECT * FROM {table} WHERE project_id = %s ORDER BY {order}", (project_id,)
|
||||
)
|
||||
|
||||
old_net, new_net = (
|
||||
rows("plan_networks", PLAN_ID, "project_id"),
|
||||
rows("plan_networks", new_id, "project_id"),
|
||||
old, now = (
|
||||
rows("plan_route_years", PLAN_ID, "year"),
|
||||
rows("plan_route_years", new_id, "year"),
|
||||
)
|
||||
assert json.loads(new_net[0]["network_json"]) == json.loads(old_net[0]["network_json"])
|
||||
for table, order in (("plan_route_pieces", "piece_id"), ("plan_route_years", "year")):
|
||||
old, now = rows(table, PLAN_ID, order), rows(table, new_id, order)
|
||||
drop = ("project_id", "updated_at")
|
||||
assert [{k: v for k, v in r.items() if k not in drop} for r in now] == [
|
||||
{k: v for k, v in r.items() if k not in drop} for r in old
|
||||
]
|
||||
drop = ("project_id", "updated_at")
|
||||
assert [{k: v for k, v in r.items() if k not in drop} for r in now] == [
|
||||
{k: v for k, v in r.items() if k not in drop} for r in old
|
||||
]
|
||||
old_a = rows("plan_route_assignees", PLAN_ID, "id")
|
||||
new_a = rows("plan_route_assignees", new_id, "id")
|
||||
assert not {r["id"] for r in old_a} & {r["id"] for r in new_a}
|
||||
assert [(r["piece_id"], r["year"], r["company_name"]) for r in new_a] == [
|
||||
(r["piece_id"], r["year"], r["company_name"]) for r in old_a
|
||||
assert [(r["piece_id"], r["company_name"]) for r in new_a] == [
|
||||
(r["piece_id"], r["company_name"]) for r in old_a
|
||||
]
|
||||
assert all(r["linked_project_id"] is None for r in new_a)
|
||||
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""590-1 예정노선 설계 내용 = 프로젝트 폴더 한 파일(요구 452 · 453) — 지킴 시험.
|
||||
|
||||
- 옮김 = 옛 DB 칸(지점 points · starts/ends · start/end · 빈 칸 = 연도 행 접기 · 노선 정본 ·
|
||||
옛 연도 노선) → 새 꼴 한 벌(좌표계 머리 한 곳 · 빈 연도 없음)
|
||||
- 파일 없음 = 빈 설계 · 옛 DB 칸(590-3 에서 옮김 끝) 안 읽음 · 코드에 옛 칸 이름 0
|
||||
- 원자 쓰기 = 바꿔치기 실패에도 옛 파일 그대로 · 임시 파일 안 남음 · 새 판 파일 거절
|
||||
- 연도 읽기 = 업체 줄 연도는 파일 조각에서(낡은 DB 연도 무시 · ⑤) · 확정 = 지문 같을 때만
|
||||
- 잠금 = 같은 프로젝트 [저장] 둘이 겹치지 않음
|
||||
@@ -24,7 +23,6 @@ sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
from helper_plan_file import use_plan, year_line_plan # noqa: E402
|
||||
|
||||
from common_util.common_util_crs import lonlat_to_tm # noqa: E402
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net # noqa: E402
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store # noqa: E402
|
||||
from D04_PlanConfirm import D04_PlanConfirm_Router as d04 # noqa: E402
|
||||
@@ -43,50 +41,59 @@ NETWORK = {
|
||||
}
|
||||
|
||||
|
||||
# --- 옮김 ---
|
||||
# --- 옛 DB 칸 안 읽음(590-3) ---
|
||||
|
||||
OLD_NAMES = ("plan_base_json", "plan_networks", "plan_route_pieces", "route_json", "start_x")
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"base",
|
||||
[
|
||||
{"epsg": 5186, "points": [A, B]},
|
||||
{"epsg": 5186, "starts": [A], "ends": [B]},
|
||||
{"epsg": 5186, "start": A, "end": B},
|
||||
],
|
||||
)
|
||||
def test_migrate_old_point_forms(base):
|
||||
plan = store.migrate(json.dumps(base), json.dumps(NETWORK), [])
|
||||
assert plan["points"] == [A, B] and plan["epsg"] == 5186
|
||||
assert plan["lines"] == NETWORK["lines"] and plan["pieces"] == NETWORK["pieces"]
|
||||
assert set(plan) == {"version", "epsg", "points", "lines", "marks", "pieces"}
|
||||
class _Pool:
|
||||
"""가짜 DB — 프로젝트 행 하나 · 부른 SQL 기록."""
|
||||
|
||||
def __init__(self):
|
||||
self.sqls = []
|
||||
|
||||
def acquire(self):
|
||||
return self
|
||||
|
||||
def cursor(self, *_):
|
||||
return self
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *_):
|
||||
return False
|
||||
|
||||
async def execute(self, sql, args=()):
|
||||
self.sqls.append(sql)
|
||||
|
||||
async def fetchone(self):
|
||||
return {"project_kind": "PLAN", "project_year": 2027, "storage_path": "x"}
|
||||
|
||||
|
||||
def test_migrate_folds_year_rows_and_legacy_route():
|
||||
rows = [
|
||||
{"year": 2027, "epsg": 5186, "route_json": None,
|
||||
"start_x": 1.0, "start_y": 2.0, "end_x": 3.0, "end_y": 4.0},
|
||||
{"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]}),
|
||||
"start_x": 0.0, "start_y": 0.0, "end_x": 1.0, "end_y": 2.0},
|
||||
{"year": 2028, "epsg": None, "route_json": None,
|
||||
"start_x": None, "start_y": None, "end_x": None, "end_y": None}, # 빈 연도 줄
|
||||
] # fmt: skip
|
||||
plan = store.migrate(None, None, rows)
|
||||
assert plan["points"] == [{"x": 0.0, "y": 0.0}, {"x": 3.0, "y": 4.0}] # 첫 시점 · 끝 종점
|
||||
assert [p["year"] for p in plan["pieces"]] == [2026] # 빈 연도는 조각이 아님
|
||||
assert store.plan_units(plan)[0]["length_m"] == 50.0
|
||||
assert store.migrate(None, None, []) == store.empty_plan()
|
||||
def test_no_file_is_empty_plan(tmp_path, monkeypatch):
|
||||
pool = _Pool()
|
||||
monkeypatch.setattr(store, "get_db_pool", lambda: pool)
|
||||
monkeypatch.setattr(store, "resolve_stored_project_path", lambda _p: str(tmp_path))
|
||||
root, plan = asyncio.run(store.load_plan(PID))
|
||||
assert root == tmp_path and plan == store.empty_plan()
|
||||
assert len(pool.sqls) == 1 and not any(n in pool.sqls[0] for n in OLD_NAMES)
|
||||
assert not store.plan_path(tmp_path).exists() # 읽기만 = 안 씀
|
||||
|
||||
|
||||
def test_migrate_one_epsg_converts_points():
|
||||
lon_lat = (127.1, 37.5)
|
||||
x, y = lonlat_to_tm(*lon_lat, 5185)
|
||||
plan = store.migrate(
|
||||
json.dumps({"epsg": 5185, "points": [{"x": x, "y": y}]}), json.dumps(NETWORK), []
|
||||
)
|
||||
assert plan["epsg"] == 5186
|
||||
assert (plan["points"][0]["x"], plan["points"][0]["y"]) == pytest.approx(
|
||||
lonlat_to_tm(*lon_lat, 5186), abs=1e-3
|
||||
)
|
||||
def test_code_has_no_old_db_names():
|
||||
"""옛 설계 칸 · 표 이름 = 코드 0(지움 SQL 025 · 지난 SQL · 시험 빼고)."""
|
||||
dirs = [p for p in ROOT.iterdir() if p.is_dir() and p.name[:3] in {"B01", "D01", "D03", "D04"}]
|
||||
files = [f for d in dirs for f in d.rglob("*") if f.suffix in {".py", ".ts"}]
|
||||
files += [*(ROOT / "common_util").glob("*.py"), *(ROOT / "db_management").glob("*.py")]
|
||||
hits = [
|
||||
f"{f.relative_to(ROOT)}:{n}"
|
||||
for f in files
|
||||
for n in OLD_NAMES
|
||||
if n in f.read_text(encoding="utf-8")
|
||||
]
|
||||
assert not hits, hits
|
||||
|
||||
|
||||
# --- 원자 쓰기 · 판 ---
|
||||
|
||||
@@ -4,12 +4,11 @@
|
||||
- B08 한 줄 주소(도로명 먼저)는 그대로
|
||||
- GET /api/map/address — 출처 VWorld
|
||||
- 등록 몸 — 기준점 필수 · 경위도 범위 · 설계속도 칸 없음
|
||||
- 옛 PLAN 기준점 = 지점 상자 가운데(꼴 셋) · 행 만들기 = 기준점 칸 셋 · 설계속도 칸 안 씀
|
||||
- 행 만들기 = 기준점 칸 셋 · 설계속도 칸 안 씀
|
||||
- 화면 — 지도 · 기준점 필수 · 출처 표기 · 설계속도 글 없음
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
@@ -25,16 +24,6 @@ from B01_Dashboard import B01_Dashboard_Map as vmap # noqa: E402
|
||||
from common_util import common_util_map_router as map_router # noqa: E402
|
||||
from D01_PlanRegister import D01_PlanRegister_Router as d01 # noqa: E402
|
||||
|
||||
|
||||
def _fill_tool():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"fill024", ROOT / "db_management" / "024_plan_site_fill.py"
|
||||
)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
PARCEL = {"type": "parcel", "text": "경상북도 울진군 금강송면 광회리 산 47",
|
||||
"structure": {"level1": "경상북도", "level2": "울진군", "level3": "",
|
||||
"level4L": "금강송면", "level4A": ""}} # fmt: skip
|
||||
@@ -80,18 +69,6 @@ def test_등록_몸_기준점_필수():
|
||||
assert (ok.site.lon, ok.site.lat) == (129.1, 36.95)
|
||||
|
||||
|
||||
def test_옛_기준점_지점_가운데():
|
||||
center = _fill_tool().base_center
|
||||
assert center({"epsg": 5186, "start": {"x": 0, "y": 10}, "end": {"x": 100, "y": 30}}) == (
|
||||
50,
|
||||
20,
|
||||
)
|
||||
pts = [{"x": 0, "y": 0}, {"x": 40, "y": 10}, {"x": 20, "y": 100}]
|
||||
assert center({"points": pts}) == (20, 50)
|
||||
assert center({"starts": [{"x": 0, "y": 0}], "ends": [{"x": 10, "y": 10}]}) == (5, 5)
|
||||
assert center(None) is None and center({"points": []}) is None
|
||||
|
||||
|
||||
def test_행_만들기_기준점_칸():
|
||||
src = (ROOT / "common_util" / "common_util_project_kind.py").read_text(encoding="utf-8")
|
||||
assert "site_lon, site_lat, site_address" in src and "design_speed_kph" not in src
|
||||
|
||||
@@ -79,9 +79,8 @@ class _Pool:
|
||||
|
||||
@pytest.fixture
|
||||
def api(monkeypatch, tmp_path):
|
||||
"""파일 = tmp_path 의 진짜 파일 · 없으면 옛 DB 칸(`base_json`)을 옮김 · 연도 상태 줄은 안 씀."""
|
||||
"""파일 = tmp_path 의 진짜 파일 · 없으면 빈 설계 · 연도 상태 줄은 안 씀."""
|
||||
state = {
|
||||
"base_json": None,
|
||||
"root": tmp_path,
|
||||
"site": SITE,
|
||||
"session": {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1},
|
||||
@@ -89,7 +88,7 @@ def api(monkeypatch, tmp_path):
|
||||
|
||||
async def load(project_id):
|
||||
plan = store.read_plan(tmp_path)
|
||||
return tmp_path, plan or store.migrate(state["base_json"], None, [])
|
||||
return tmp_path, plan or store.empty_plan()
|
||||
|
||||
async def project_info(project_id):
|
||||
return {"year_start": 2026, "site": state["site"]}
|
||||
@@ -159,15 +158,6 @@ def test_epsg_stays_once_saved(api) -> None:
|
||||
assert got.status_code == 400 # 파일 머리 좌표계 한 곳
|
||||
|
||||
|
||||
def test_old_start_end_points_visible(api) -> None:
|
||||
old = {"epsg": 5186, "starts": [{"x": A[0], "y": A[1]}], "ends": [{"x": B[0], "y": B[1]}]}
|
||||
api.state["base_json"] = json.dumps(old)
|
||||
points = api.get(URL).json()["base"]["points"]
|
||||
assert [p["x"] for p in points] == pytest.approx([A[0], B[0]]) # 시점 뒤 종점
|
||||
api.state["base_json"] = json.dumps({"epsg": 5186, "start": {"x": A[0], "y": A[1]}})
|
||||
assert [p["x"] for p in api.get(URL).json()["base"]["points"]] == pytest.approx([A[0]])
|
||||
|
||||
|
||||
# --- D02 걷음 ---
|
||||
|
||||
|
||||
|
||||
@@ -118,7 +118,7 @@ def test_D04_확정_시각_끝_Z():
|
||||
from D04_PlanConfirm.D04_PlanConfirm_Router import unit_summary, year_summary
|
||||
|
||||
at = datetime(2026, 10, 30, 1, 2, 3)
|
||||
assert year_summary({"year": 2026, "route_json": {}, "confirmed_at": at})["confirmed_at"] == (
|
||||
assert year_summary({"year": 2026, "route": {}, "confirmed_at": at})["confirmed_at"] == (
|
||||
"2026-10-30T01:02:03Z"
|
||||
)
|
||||
unit = {
|
||||
|
||||
Reference in New Issue
Block a user