eomsangdonandClaude Opus 5.5
f5645c256d
- 기준지정 시점 · 종점 목록 → 「지점 목록」 하나 · [+ 지점] · 지점마다 차례 색 여섯 돌림(줄 왼 띠 · 지도 이름표 바탕)
- 저장 꼴 plan_base_json {epsg, points} · 옛 starts + ends · start · end 는 읽을 때 이어 붙임(옮김 없음) · 상한 40
- 노선분석 지점 = 핀 + 「지점 N」 · 기준지정과 같은 색 · 겹 목록 맨 끝(늘 맨 위) · 지점 누름 = 새 본선(No.0 = 그 지점) · 선 놓인 지점 = 갈래
- 공용 카카오 지도 = 휠 설정 방향 한 단(마우스 자리 고정) · 가운데 끌기 옮김 — 기준지정 · 지형 범위 창 · B08 이정표 같이
- 시험: 529-3 · 530-2 새 꼴 · 585-1 새 지킴 시험 · 낡은 B08 가짜 카카오(바탕 종류 · 사건 받기 없음) 맞춤
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
434 lines
17 KiB
Python
434 lines
17 KiB
Python
"""D 예정노선 2단계 노선 정본 — 손으로 그은 직선 · 갈래 · 구역
|
|
(요구 389 6 · 11 · 13 · 14 · 391 · PLAN 562-2).
|
|
|
|
- `GET /api/projects/{id}/plan/network` — 정본(`plan_networks`) + 서버가 다시 센 길이 ·
|
|
측점 · 조각 · 저장 없으면 옛 연도 노선(`plan_route_years.route_json.planned`)을 한 선으로 바꿔
|
|
보임(legacy)
|
|
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
|
|
다시 셈 → `plan_networks` · 조각마다 한 줄(`plan_route_pieces`) · 연도 행(`plan_route_years`) ·
|
|
사업연도 끝
|
|
|
|
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
|
|
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
|
|
지점 목록(`base` · 585-1) = D02 정본(`plan_base.load_base`) — 비어 옛 연도 행을 접어 보이던 것은
|
|
[저장] 전에 칸에 굳힘(연도 행이 바뀌어도 그대로).
|
|
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from common_util.common_util_auth import is_forest_service, require_company
|
|
from config.config_db import get_db_pool
|
|
from D02_PlanBase import D02_PlanBase_Router as plan_base
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Draw as draw
|
|
|
|
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Network"])
|
|
|
|
ID = r"^[A-Za-z0-9_-]{1,64}$"
|
|
SAME_M = 0.001 # 1 mm 안 = 같은 점 · 같은 마크
|
|
PIECE_TOL_M = 0.01 # 조각 끝 = 마크 경계(1 cm)
|
|
STATION_M = 20.0 # 측점 간격(지식DB 설계제원_총괄)
|
|
|
|
|
|
class Branch(BaseModel):
|
|
line: str = Field(pattern=ID)
|
|
vertex: int = Field(ge=0)
|
|
|
|
|
|
class Line(BaseModel):
|
|
id: str = Field(pattern=ID)
|
|
vertices: list[tuple[float, float]] = Field(min_length=1)
|
|
branch: Branch | None = None
|
|
|
|
|
|
class Mark(BaseModel):
|
|
line: str = Field(pattern=ID)
|
|
at_m: float
|
|
|
|
|
|
class Piece(BaseModel):
|
|
id: str = Field(pattern=ID)
|
|
line: str = Field(pattern=ID)
|
|
from_m: float
|
|
to_m: float
|
|
year: int
|
|
|
|
|
|
class Network(BaseModel):
|
|
epsg: int
|
|
lines: list[Line] = []
|
|
marks: list[Mark] = []
|
|
pieces: list[Piece] = []
|
|
|
|
|
|
class NetworkError(ValueError):
|
|
pass
|
|
|
|
|
|
def _check(ok: bool, message: str) -> None:
|
|
if not ok:
|
|
raise NetworkError(message)
|
|
|
|
|
|
def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, Any]:
|
|
"""선 · 마크 · 조각 → 선 길이 · 측점 · 조각(경계 = 마크 · 꺾은선 · 측점) ·
|
|
틀리면 `NetworkError`.
|
|
|
|
`year_from` = 사업연도 시작(그 앞 연도 조각 거절 · None = 안 봄)."""
|
|
lines: dict[str, list] = {}
|
|
for line in network["lines"]:
|
|
vertices = line["vertices"]
|
|
_check(line["id"] not in lines, f"선 id 가 겹칩니다: {line['id']}")
|
|
_check(
|
|
all(math.isfinite(v) for point in vertices for v in point),
|
|
f"선 {line['id']} 점이 수가 아닙니다.",
|
|
)
|
|
branch = line.get("branch")
|
|
if branch:
|
|
parent = lines.get(branch["line"])
|
|
_check(parent is not None, f"갈래 {line['id']} 의 부모 선이 앞에 없습니다.")
|
|
_check(branch["vertex"] < len(parent), f"갈래 {line['id']} 의 부모 점이 없습니다.")
|
|
_check(
|
|
draw.segment_length(vertices[0], parent[branch["vertex"]]) <= SAME_M,
|
|
f"갈래 {line['id']} 첫 점이 부모 선 점과 다릅니다.",
|
|
)
|
|
lines[line["id"]] = vertices
|
|
lengths = {key: draw.line_length(vertices) for key, vertices in lines.items()}
|
|
cuts: dict[str, list[float]] = {key: [] for key in lines}
|
|
for mark in network["marks"]:
|
|
_check(mark["line"] in lines, f"마크의 선이 없습니다: {mark['line']}")
|
|
at = mark["at_m"]
|
|
_check(
|
|
math.isfinite(at) and SAME_M < at < lengths[mark["line"]] - SAME_M,
|
|
f"마크가 선 {mark['line']} 밖입니다: {at}",
|
|
)
|
|
cuts[mark["line"]].append(at)
|
|
tiles: dict[str, list[tuple[float, float]]] = {}
|
|
for key, marks in cuts.items():
|
|
edges = [0.0, *sorted(marks), lengths[key]]
|
|
_check(
|
|
all(b - a > SAME_M for a, b in zip(edges[1:-1], edges[2:-1])),
|
|
f"선 {key} 마크가 겹칩니다.",
|
|
)
|
|
tiles[key] = list(zip(edges, edges[1:])) if lengths[key] > SAME_M else []
|
|
order = {key: i for i, key in enumerate(lines)}
|
|
stations = {key: draw.station_chainages(lengths[key], STATION_M) for key in lines}
|
|
used: set[tuple[str, int]] = set()
|
|
pieces = []
|
|
for piece in network["pieces"]:
|
|
key = piece["line"]
|
|
_check(key in lines, f"조각 {piece['id']} 의 선이 없습니다.")
|
|
_check(year_from is None or piece["year"] >= year_from, f"조각 연도가 사업연도 앞: {piece}")
|
|
hit = next(
|
|
(
|
|
i
|
|
for i, (a, b) in enumerate(tiles[key])
|
|
if abs(a - piece["from_m"]) <= PIECE_TOL_M and abs(b - piece["to_m"]) <= PIECE_TOL_M
|
|
),
|
|
None,
|
|
)
|
|
_check(hit is not None, f"조각 {piece['id']} 끝이 마크 경계와 다릅니다.")
|
|
_check((key, hit) not in used, f"조각이 겹칩니다: {piece['id']}")
|
|
used.add((key, hit))
|
|
a, b = tiles[key][hit]
|
|
pieces.append(
|
|
{
|
|
"id": piece["id"],
|
|
"line": key,
|
|
"year": int(piece["year"]),
|
|
"from_m": a,
|
|
"to_m": b,
|
|
"length_m": b - a,
|
|
"planned": [list(p) for p in draw.slice_line(lines[key], a, b)],
|
|
"stations": sorted(
|
|
{a, b, *(c for c in stations[key] if a - 1e-6 <= c <= b + 1e-6)}
|
|
),
|
|
}
|
|
)
|
|
_check(len({p["id"] for p in pieces}) == len(pieces), "조각 id 가 겹칩니다.")
|
|
pieces.sort(key=lambda p: (order[p["line"]], p["from_m"]))
|
|
return {
|
|
"lines": [
|
|
{"id": key, "length_m": lengths[key], "stations": stations[key]} for key in lines
|
|
],
|
|
"pieces": pieces,
|
|
}
|
|
|
|
|
|
def year_rows(pieces: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
|
|
"""조각 → 연도 행(`plan_route_years`) — `planned` = 첫 조각(옛 읽는 곳 그대로) ·
|
|
`pieces` = 그 연도 전부."""
|
|
by_year: dict[int, list[dict[str, Any]]] = {}
|
|
for piece in pieces:
|
|
by_year.setdefault(piece["year"], []).append(piece)
|
|
out = {}
|
|
for year, items in sorted(by_year.items()):
|
|
first = items[0]
|
|
parts = [
|
|
{
|
|
"id": p["id"],
|
|
"line": p["line"],
|
|
"planned": p["planned"],
|
|
"start_chainage_m": p["from_m"],
|
|
"end_chainage_m": p["to_m"],
|
|
"length_m": p["length_m"],
|
|
}
|
|
for p in items
|
|
]
|
|
route = {
|
|
"mode": "manual",
|
|
"planned": first["planned"],
|
|
"start_chainage_m": first["from_m"],
|
|
"end_chainage_m": first["to_m"],
|
|
"length_m": sum(p["length_m"] for p in items),
|
|
"pieces": parts,
|
|
}
|
|
out[year] = {
|
|
"route": route,
|
|
"fingerprint": _sha1(parts),
|
|
"start": first["planned"][0],
|
|
"end": first["planned"][-1],
|
|
}
|
|
return out
|
|
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
async def _load(connection: Any, project_id: UUID) -> tuple[dict[str, Any] | None, bool]:
|
|
"""(정본 · 옛 연도 노선을 바꾼 것인가) — 둘 다 없으면 (None, False)."""
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"SELECT network_json FROM plan_networks WHERE project_id = %s", (str(project_id),)
|
|
)
|
|
row = await cursor.fetchone()
|
|
if row:
|
|
return _json(row["network_json"]), False
|
|
await cursor.execute(
|
|
"SELECT year, epsg, route_json FROM plan_route_years"
|
|
" WHERE project_id = %s ORDER BY year",
|
|
(str(project_id),),
|
|
)
|
|
legacy = legacy_network(list(await cursor.fetchall()))
|
|
return legacy, legacy is not None
|
|
|
|
|
|
def _view(
|
|
network: dict[str, Any] | None, legacy: bool, base: dict[str, Any], year_start: int | None
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"status": "success",
|
|
"base": base,
|
|
"year_start": year_start,
|
|
"network": network,
|
|
"derived": derive(network) if network else None,
|
|
"legacy": legacy,
|
|
}
|
|
|
|
|
|
@router.get("/{project_id}/plan/network", response_model=None)
|
|
async def get_plan_network(project_id: UUID) -> dict[str, Any]:
|
|
base = await plan_base.load_base(project_id)
|
|
async with get_db_pool().acquire() as connection:
|
|
network, legacy = await _load(connection, project_id)
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"SELECT project_year FROM projects WHERE id = %s", (str(project_id),)
|
|
)
|
|
row = await cursor.fetchone()
|
|
try:
|
|
return _view(network, legacy, base, row["project_year"] if row else None)
|
|
except NetworkError as error:
|
|
raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
|
|
|
|
|
|
async def _store(
|
|
connection: Any, project_id: UUID, network: dict[str, Any], derived: dict[str, Any]
|
|
) -> None:
|
|
pid = str(project_id)
|
|
epsg = network["epsg"]
|
|
years = year_rows(derived["pieces"])
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT DISTINCT piece_id FROM plan_route_assignees"
|
|
" WHERE project_id = %s AND piece_id IS NOT NULL",
|
|
(pid,),
|
|
)
|
|
assigned = {r[0] for r in await cursor.fetchall()}
|
|
await cursor.execute(
|
|
"SELECT DISTINCT year FROM plan_route_assignees"
|
|
" WHERE project_id = %s AND piece_id IS NULL",
|
|
(pid,),
|
|
)
|
|
assigned_years = {int(r[0]) for r in await cursor.fetchall()}
|
|
lost = sorted(assigned - {p["id"] for p in derived["pieces"]})
|
|
lost_years = sorted(assigned_years - set(years))
|
|
if lost or lost_years:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"업체가 지정된 조각 · 연도를 지울 수 없습니다: {lost + lost_years}",
|
|
)
|
|
await cursor.execute(
|
|
"INSERT INTO plan_networks (project_id, network_json) VALUES (%s, %s)"
|
|
" ON DUPLICATE KEY UPDATE network_json = VALUES(network_json)",
|
|
(pid, json.dumps(network, ensure_ascii=False)),
|
|
)
|
|
await cursor.execute("DELETE FROM plan_route_pieces WHERE project_id = %s", (pid,))
|
|
for p in derived["pieces"]:
|
|
route = {"planned": p["planned"], "stations": p["stations"]}
|
|
await cursor.execute(
|
|
"INSERT INTO plan_route_pieces (project_id, piece_id, line_id, year, from_m, to_m,"
|
|
" length_m, epsg, route_json, fingerprint)"
|
|
" VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
|
|
(
|
|
pid,
|
|
p["id"],
|
|
p["line"],
|
|
p["year"],
|
|
p["from_m"],
|
|
p["to_m"],
|
|
p["length_m"],
|
|
epsg,
|
|
json.dumps(route),
|
|
_sha1(route),
|
|
),
|
|
)
|
|
for year, row in years.items():
|
|
# 내용이 바뀐 연도만 확정 풂(confirmed_at 을 fingerprint 보다 먼저 셈)
|
|
await cursor.execute(
|
|
"INSERT INTO plan_route_years (project_id, year, start_x, start_y, end_x, end_y,"
|
|
" epsg, route_json, fingerprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
|
" ON DUPLICATE KEY UPDATE start_x = VALUES(start_x), start_y = VALUES(start_y),"
|
|
" end_x = VALUES(end_x), end_y = VALUES(end_y), epsg = VALUES(epsg),"
|
|
" route_json = VALUES(route_json),"
|
|
" confirmed_at = IF(fingerprint <=> VALUES(fingerprint), confirmed_at, NULL),"
|
|
" fingerprint = VALUES(fingerprint)",
|
|
(
|
|
pid,
|
|
year,
|
|
*row["start"],
|
|
*row["end"],
|
|
epsg,
|
|
json.dumps(row["route"], ensure_ascii=False),
|
|
row["fingerprint"],
|
|
),
|
|
)
|
|
if years:
|
|
marks = ", ".join(["%s"] * len(years))
|
|
await cursor.execute(
|
|
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
|
|
f" confirmed_at = NULL WHERE project_id = %s AND year NOT IN ({marks})",
|
|
(pid, *years),
|
|
)
|
|
await cursor.execute(
|
|
"UPDATE projects SET project_year_end = %s WHERE id = %s", (max(years), pid)
|
|
)
|
|
else:
|
|
await cursor.execute(
|
|
"UPDATE plan_route_years SET route_json = NULL, fingerprint = NULL,"
|
|
" confirmed_at = NULL WHERE project_id = %s",
|
|
(pid,),
|
|
)
|
|
|
|
|
|
async def _freeze_base(connection: Any, project_id: UUID, base: dict[str, Any]) -> None:
|
|
"""옛 연도 행을 접은 지점 → D02 정본 칸(비었을 때만)."""
|
|
stored = {"epsg": base["epsg"], "points": [{"x": p["x"], "y": p["y"]} for p in base["points"]]}
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"UPDATE projects SET plan_base_json = %s WHERE id = %s AND plan_base_json IS NULL",
|
|
(json.dumps(stored), str(project_id)),
|
|
)
|
|
|
|
|
|
@router.put("/{project_id}/plan/network", response_model=None)
|
|
async def put_plan_network(
|
|
project_id: UUID,
|
|
payload: Network,
|
|
session: dict[str, Any] = Depends(require_company),
|
|
) -> dict[str, Any]:
|
|
"""[저장] · [초기화](lines 빈 것) — 서버가 정본으로 다시 셈 → 정본 · 조각 행 · 연도 행."""
|
|
if not is_forest_service(session):
|
|
raise HTTPException(status_code=403, detail="예정노선은 산림청만 고칠 수 있습니다.")
|
|
network = payload.model_dump()
|
|
base = await plan_base.load_base(project_id) # 예정노선 프로젝트가 아니면 404 · 400
|
|
if base["epsg"] is not None and int(base["epsg"]) != network["epsg"]:
|
|
raise HTTPException(status_code=400, detail="좌표계가 기준 지정과 다릅니다.")
|
|
async with get_db_pool().acquire() as connection:
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"SELECT project_year, plan_base_json FROM projects WHERE id = %s",
|
|
(str(project_id),),
|
|
)
|
|
project = await cursor.fetchone()
|
|
try:
|
|
derived = derive(network, project["project_year"])
|
|
except NetworkError as error:
|
|
raise HTTPException(status_code=400, detail=str(error)) from error
|
|
await connection.begin()
|
|
try:
|
|
if project["plan_base_json"] is None and base["points"]:
|
|
await _freeze_base(connection, project_id, base)
|
|
await _store(connection, project_id, network, derived)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return {**_view(network, False, base, project["project_year"]), "derived": derived}
|