eomsangdon
b92ba4592e
사용자 지시로 두 값을 걷어낸다. 구조물별 옵션·정의는 앞으로 타입마다 따로 정한다. 어차피 소비처가 없어(B06~B08은 structures를 읽지 않는다) 화면에만 남아 있던 값이었다. - 폼: 설치측 select·이격 칸 행 삭제, 목록 부제·종단 벌룬에서 측 표기 제거 - 계약: StructureSide 타입, StructureInstance.side/offset_m 삭제(프론트·스키마) - 설계지문에서 두 값 제외 — 후속 단계 무효화 판정에서 빠진다 - 구 저장분은 읽을 때 두 키만 떨어낸다. extra="forbid"라 그냥 두면 정본이 통째로 버려진다. 다른 낯선 키는 계속 거절한다
211 lines
8.8 KiB
Python
211 lines
8.8 KiB
Python
"""B05 구조물 정본(`B05_Profile/route/structures.json`) 읽기·쓰기.
|
|
|
|
정본은 이 파일 **하나뿐**이다. DB에는 참조 메타(개수·revision)만 남긴다 — 같은 값을 두 곳에
|
|
두면 한쪽 저장이 실패했을 때 어느 쪽이 진짜인지 알 수 없다(관 매설 지점의 "복원 유령" 전례).
|
|
|
|
저장은 임시 파일에 쓰고 교체하는 방식이라, 쓰는 도중 죽어도 반쪽짜리 파일이 남지 않는다.
|
|
동시에 두 화면이 저장하면 `base_revision`이 어긋나 뒤엣것이 거절된다(앞의 편집을 덮지 않게).
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
from typing import Any, Iterable
|
|
|
|
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance, structure_type_map
|
|
from common_util.common_util_json import atomic_write_json
|
|
|
|
STRUCTURES_FILE_NAME = "structures.json"
|
|
_STAGE_DIR = os.path.join("B05_Profile", "route")
|
|
|
|
|
|
class StructureRevisionConflict(RuntimeError):
|
|
"""다른 저장이 먼저 반영되어 판번호가 어긋났다."""
|
|
|
|
def __init__(self, expected: int, actual: int) -> None:
|
|
super().__init__(
|
|
f"구조물 정본이 이미 갱신되었습니다 (요청 판번호 {expected}, 현재 {actual}). "
|
|
"최신 내용을 다시 불러온 뒤 저장해 주세요."
|
|
)
|
|
self.expected = expected
|
|
self.actual = actual
|
|
|
|
|
|
# 2026-08-17 사용자 지시로 걷어낸 필드 — 설치측·이격은 받지 않는다. 스키마가
|
|
# extra="forbid"라 남아 있는 저장분을 그대로 넘기면 정본 전체가 통째로 버려지므로,
|
|
# 읽을 때 이 두 키만 떨어낸다. 다른 낯선 키는 계속 거절해 정본이 조용히 썩는 것을 막는다.
|
|
_DROPPED_KEYS = ("side", "offset_m")
|
|
|
|
|
|
def structures_file_path(project_root: str) -> str:
|
|
"""프로젝트 저장소 안의 구조물 정본 경로."""
|
|
return os.path.join(project_root, _STAGE_DIR, STRUCTURES_FILE_NAME)
|
|
|
|
|
|
def load_structures(project_root: str) -> tuple[int, list[StructureInstance]]:
|
|
"""정본을 읽어 (판번호, 구조물 목록)을 돌려준다.
|
|
|
|
파일이 없거나 깨졌으면 빈 정본(판번호 0)으로 본다 — 화면이 못 열리는 것보다 낫고,
|
|
다음 저장이 온전한 파일로 덮어쓴다.
|
|
"""
|
|
path = structures_file_path(project_root)
|
|
if not os.path.exists(path):
|
|
return 0, []
|
|
try:
|
|
with open(path, encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
revision = int(payload.get("revision", 0))
|
|
structures = [
|
|
StructureInstance.model_validate(_without_dropped_keys(item))
|
|
for item in payload.get("structures", [])
|
|
]
|
|
except (OSError, ValueError, TypeError):
|
|
return 0, []
|
|
return revision, structures
|
|
|
|
|
|
def _without_dropped_keys(item: Any) -> Any:
|
|
"""폐지된 필드가 남아 있는 저장분을 지금 스키마로 읽을 수 있게 손질한다."""
|
|
if not isinstance(item, dict):
|
|
return item
|
|
if not any(key in item for key in _DROPPED_KEYS):
|
|
return item
|
|
return {key: value for key, value in item.items() if key not in _DROPPED_KEYS}
|
|
|
|
|
|
def save_structures(
|
|
project_root: str,
|
|
structures: Iterable[StructureInstance],
|
|
*,
|
|
base_revision: int,
|
|
max_chainage_m: float | None = None,
|
|
) -> int:
|
|
"""구조물 목록을 정본에 덮어쓰고 새 판번호를 돌려준다.
|
|
|
|
`max_chainage_m`는 노선 총연장(m) — 주어지면 범위 밖 배치를 거절한다.
|
|
"""
|
|
items = list(structures)
|
|
_validate_types(items)
|
|
_validate_unique_ids(items)
|
|
_validate_range(items, max_chainage_m)
|
|
|
|
current_revision, _ = load_structures(project_root)
|
|
if current_revision != base_revision:
|
|
raise StructureRevisionConflict(base_revision, current_revision)
|
|
|
|
for item in items:
|
|
if not item.structure_id:
|
|
item.structure_id = uuid.uuid4().hex
|
|
|
|
revision = current_revision + 1
|
|
payload = {
|
|
"revision": revision,
|
|
"structures": [item.model_dump(mode="json") for item in items],
|
|
}
|
|
atomic_write_json(structures_file_path(project_root), payload)
|
|
return revision
|
|
|
|
|
|
def requires_downstream_invalidation(
|
|
previous: Iterable[StructureInstance], current: Iterable[StructureInstance]
|
|
) -> bool:
|
|
"""구조물 변경이 B06 이후 결과를 못 쓰게 만드는지 판정한다.
|
|
|
|
메모나 표시용 값만 바뀐 경우까지 후속 단계를 깨면, 사용자가 메모 한 줄 고칠 때마다
|
|
횡단·수량을 다시 돌려야 한다. 그래서 설계에 실제로 영향을 주는 값
|
|
(타입·위치·범위·제원)만 비교한다. 목록 순서는 설계와 무관하므로 무시한다.
|
|
"""
|
|
return _design_fingerprint(previous) != _design_fingerprint(current)
|
|
|
|
|
|
def _design_fingerprint(items: Iterable[StructureInstance]) -> set[str]:
|
|
return {
|
|
json.dumps(
|
|
[
|
|
item.structure_id,
|
|
item.type_id,
|
|
item.placement,
|
|
item.chainage_m,
|
|
item.start_m,
|
|
item.end_m,
|
|
item.options,
|
|
item.geometry,
|
|
],
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
)
|
|
for item in items
|
|
}
|
|
|
|
|
|
def _validate_types(items: list[StructureInstance]) -> None:
|
|
"""레지스트리 대조 검증 — 타입 존재·관리 주체·배치형태·옵션까지 서버가 지킨다.
|
|
|
|
화면만 믿으면 조작된 요청(placement 불일치·음수 제원·미정의 옵션)이 정본에
|
|
들어간다(2026-08-16 크로스체크 지적 2). 정본에 닿는 마지막 관문은 여기다.
|
|
"""
|
|
types = structure_type_map()
|
|
for item in items:
|
|
definition = types.get(item.type_id)
|
|
if definition is None:
|
|
raise ValueError(f"등록되지 않은 구조물 타입입니다: {item.type_id}")
|
|
if definition.managed_by:
|
|
raise ValueError(
|
|
f"{definition.name}은(는) {definition.managed_by} 정본이 관리합니다 — "
|
|
"구조물 목록에 저장할 수 없습니다."
|
|
)
|
|
if item.placement != definition.placement:
|
|
raise ValueError(
|
|
f"{definition.name}의 배치형태는 {definition.placement}인데 "
|
|
f"{item.placement}로 보냈습니다."
|
|
)
|
|
_validate_options(item, definition)
|
|
|
|
|
|
def _validate_options(item: StructureInstance, definition) -> None:
|
|
allowed = {option.key: option for option in definition.options}
|
|
for key, value in item.options.items():
|
|
option = allowed.get(key)
|
|
if option is None:
|
|
raise ValueError(f"{definition.name}에 정의되지 않은 옵션입니다: {key}")
|
|
if option.input == "number":
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValueError(f"{definition.name}의 {option.label}은(는) 숫자여야 합니다.")
|
|
if not (value == value and abs(value) != float("inf")) or value < 0:
|
|
raise ValueError(f"{definition.name}의 {option.label}은(는) 0 이상이어야 합니다.")
|
|
elif option.input == "select" and option.choices and str(value) not in option.choices:
|
|
raise ValueError(f"{definition.name}의 {option.label} 값이 선택지에 없습니다: {value}")
|
|
for option in definition.options:
|
|
# phase=detail은 B06/B07 상세 단계 입력 — B05 저장(유무·종류·위치)에서는
|
|
# 강제하지 않는다. 필수 원칙은 유지되고 시점만 미뤄진다(2026-08-17 사용자 확정).
|
|
if option.phase == "detail":
|
|
continue
|
|
if option.required and item.options.get(option.key) in (None, ""):
|
|
raise ValueError(
|
|
f"{definition.name}의 {option.label}은(는) 필수 입력입니다 — "
|
|
"미확정 항목이라 기본값이 없습니다."
|
|
)
|
|
|
|
|
|
def _validate_unique_ids(items: list[StructureInstance]) -> None:
|
|
seen: set[str] = set()
|
|
for item in items:
|
|
if not item.structure_id:
|
|
continue
|
|
if item.structure_id in seen:
|
|
raise ValueError(f"구조물 식별자가 중복되었습니다: {item.structure_id}")
|
|
seen.add(item.structure_id)
|
|
|
|
|
|
def _validate_range(items: list[StructureInstance], max_chainage_m: float | None) -> None:
|
|
"""노선 연장을 알 때만 범위를 지킨다 — 연장 밖 배치는 도면·수량 어디에도 못 실린다."""
|
|
if max_chainage_m is None:
|
|
return
|
|
limit = max_chainage_m + 1e-6
|
|
for item in items:
|
|
positions = [item.chainage_m, item.start_m, item.end_m]
|
|
if any(value is not None and value > limit for value in positions):
|
|
raise ValueError(
|
|
f"구조물 위치가 노선 연장({max_chainage_m:.1f} m)을 벗어났습니다: {item.type_id}"
|
|
)
|