- 칸 조작은 캐시(sessionStorage)에, [저장] 때 structures.json·pipe_points.json 에 바로 씀 - 산출 조건에 손댄 칸 표(고치기 전 값) · ↺ 로 되돌리면 자동으로 돌아감 - 상세 칸만 고침 — 자리·길이·높이·관경은 시·종점과 한 벌이라 구조물 놓기(B05) 몫 - 틀린 값·놓기 칸·판번호 어긋남은 아무것도 안 씀 · 값 꼴은 구조물도 제원 저장과 같음 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq
359 lines
16 KiB
Python
359 lines
16 KiB
Python
"""B08 **구조물 집계표** — 측점별 구조물 한 줄 · 종류별 표 (PLAN 2장 · 사용자 확정 ④).
|
|
|
|
한 줄 = 측점 + 종류 + 실치수(제원 칸) + 개소·연장. 실무 「돌-골막이치수」 치수조서가
|
|
실치수를 들고 구조도 탭이 끌어 쓰는 모양을 **한 탭으로 겸함**.
|
|
|
|
값이 어디서 오나 — 칸마다 출처를 붙임
|
|
auto 정본에 적힌 값 — `structures.json`(B05) · `pipe_points.json`(계곡 통과 시설) ·
|
|
관 연장은 B06 횡단 `design.pipe_length_m`
|
|
user 이 표에서 사람이 고쳐 **정본에 적은** 값 — 산출 조건에 「손댄 칸」 표만 둠(브레인 판정)
|
|
library 정본 칸이 비어 **양식 기본값**으로 선 칸(구조물도 양식 `vars.*.default`)
|
|
empty 정본도 양식도 값이 없음 — 계산 쪽이 막히거나 기준값으로 돎
|
|
⚠ 사용자 손 값은 덮개층이 아니라 **정본에 씀** — 도면·수량이 한 값(지침 5장 「쪼개지 않음」).
|
|
⚠ 손댄 칸의 정본 값이 뒤에 B05 에서 바뀌면 그 칸은 자동으로 돌아가되 **조용히 말고** 줄에 알림.
|
|
⚠ 값을 여기서 짓지 않음 — 읽어 줄 세우기만. 연장이 없는 관은 빈칸(0 으로 안 채움).
|
|
⚠ 관 지점 정본 타입(`managed_by`)은 `structures.json` 에 있어도 안 셈 — `pipe_points.json` 이 주인.
|
|
⛔ 고칠 수 있는 칸은 **상세 칸(`phase: detail`)만** — 자리·길이·높이·관경 같은 놓기 칸(`b05`)은
|
|
시·종점·횡단 설계와 **한 벌로 움직여**(B05 에서 길이 = 전 + 후, 시·종점이 따라 섬) 여기서 한 칸만
|
|
고치면 쪼개짐(지침 5장). 그 칸은 구조물 놓기(B05)에서 고침.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from collections.abc import Iterable
|
|
from typing import Any
|
|
|
|
SOURCE_AUTO = "auto"
|
|
SOURCE_USER = "user"
|
|
SOURCE_LIBRARY = "library"
|
|
SOURCE_EMPTY = "empty"
|
|
#: 산출 조건 자리 — 손댄 칸 `{줄 id: {칸: {"value": 적은 값, "was": 고치기 전 정본 값}}}`.
|
|
USER_CELLS_KEY = "structure_summary_user_cells"
|
|
|
|
#: 계곡 통과 시설 `facility` → 레지스트리 종류. 빈 값은 배관(`PipePoint` 기본).
|
|
FACILITY_TYPES = {
|
|
"pipe": "pipe",
|
|
"box_culvert": "box_culvert",
|
|
"ford_pavement": "ford_pavement",
|
|
"ford_bridge": "ford_bridge",
|
|
"revetment": "revetment",
|
|
}
|
|
|
|
|
|
def _blank(value: Any) -> bool:
|
|
return value is None or value == ""
|
|
|
|
|
|
def _number(value: Any) -> float | None:
|
|
if isinstance(value, bool) or _blank(value):
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _defaults(template: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""양식 제원 칸 기본값 `{옵션 키: 기본값}` — 정본 칸이 비었을 때 계산이 쓰는 값."""
|
|
return {
|
|
str(spec["option"]): spec["default"]
|
|
for spec in ((template or {}).get("vars") or {}).values()
|
|
if spec.get("option") and "default" in spec and not _blank(spec["default"])
|
|
}
|
|
|
|
|
|
def _same(left: Any, right: Any) -> bool:
|
|
"""정본 값과 적은 값이 같은가 — 숫자는 3.0 과 "3" 을 같게 봄."""
|
|
a, b = _number(left), _number(right)
|
|
return a == b if a is not None and b is not None else left == right
|
|
|
|
|
|
def _cells(
|
|
keys: Iterable[str],
|
|
options: dict[str, Any],
|
|
defaults: dict[str, Any],
|
|
marks: dict[str, Any],
|
|
) -> dict:
|
|
cells = {}
|
|
for key in keys:
|
|
current = None if _blank(options.get(key)) else options[key]
|
|
mark = marks.get(key)
|
|
if mark is not None and not _same(current, mark.get("value")):
|
|
# 사람이 적은 값을 B05 가 바꿈 — 자동으로 돌아가되 옛 값을 함께 실어 화면이 알림.
|
|
cells[key] = {
|
|
"value": current,
|
|
"source": SOURCE_EMPTY if current is None else SOURCE_AUTO,
|
|
"replaced_user_value": mark.get("value"),
|
|
}
|
|
elif mark is not None:
|
|
cells[key] = {"value": current, "source": SOURCE_USER, "was": mark.get("was")}
|
|
elif current is not None:
|
|
cells[key] = {"value": options[key], "source": SOURCE_AUTO}
|
|
elif key in defaults:
|
|
cells[key] = {"value": defaults[key], "source": SOURCE_LIBRARY}
|
|
else:
|
|
cells[key] = {"value": None, "source": SOURCE_EMPTY}
|
|
return cells
|
|
|
|
|
|
def build_summary(
|
|
structures: Iterable[dict[str, Any]],
|
|
pipe_points: Iterable[dict[str, Any]],
|
|
types: dict[str, Any],
|
|
templates: dict[str, dict[str, Any]] | None = None,
|
|
pipe_lengths: dict[float, float] | None = None,
|
|
user_cells: dict[str, dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""종류별 표 목록(레지스트리 차례) — 표마다 칸 정의·줄·개소·연장 합·평균치수.
|
|
|
|
`types` 는 `structure_type_map()` · `templates` 는 종류별 양식(프로젝트에 박힌 것 → 기본) ·
|
|
`pipe_lengths` 는 B06 횡단의 `{측점: 관 연장}`(`pipe_lengths_from_designs`) ·
|
|
`user_cells` 는 이 표에서 손댄 칸(`USER_CELLS_KEY` 모양).
|
|
"""
|
|
# 관 길이 찾기는 배수관 물량과 **같은 규칙**(허용 0.5m) — 두 표가 같은 관에 다른 연장을 안 적게.
|
|
from B08_Quantity.B08_Quantity_Engine_Pipe import _nearest
|
|
|
|
templates = templates or {}
|
|
lengths = pipe_lengths or {}
|
|
rows_by_type: dict[str, list[dict[str, Any]]] = {}
|
|
notes: list[str] = []
|
|
|
|
def add(type_id: str, row: dict[str, Any]) -> None:
|
|
definition = types.get(type_id)
|
|
if definition is None:
|
|
notes.append(f"{type_id}: 레지스트리에 없는 종류라 안 셈")
|
|
return
|
|
options = row.pop("options")
|
|
keys = [field.key for field in definition.options]
|
|
marks = (user_cells or {}).get(str(row["id"])) or {}
|
|
row["cells"] = _cells(keys, options, _defaults(templates.get(type_id)), marks)
|
|
replaced = [key for key, cell in row["cells"].items() if "replaced_user_value" in cell]
|
|
if replaced:
|
|
labels = {field.key: field.label for field in definition.options}
|
|
row["replaced"] = [labels[key] for key in replaced]
|
|
if definition.design_owner:
|
|
row["note"] = f"{definition.design_owner}가 수량을 셈 — 여기선 자리만"
|
|
elif definition.reference_only:
|
|
row["note"] = "전문 상세설계 대상 — 배치까지만"
|
|
rows_by_type.setdefault(type_id, []).append(row)
|
|
|
|
for item in structures:
|
|
type_id = str(item.get("type_id") or "")
|
|
definition = types.get(type_id)
|
|
if definition is not None and definition.managed_by:
|
|
continue # 관 지점 정본이 주인 — 옛 저장분이 남아 있어도 두 번 안 셈
|
|
options = dict(item.get("options") or {})
|
|
start, end = item.get("start_m"), item.get("end_m")
|
|
span = abs(float(end) - float(start)) if start is not None and end is not None else None
|
|
stated = _number(options.get("length_m"))
|
|
add(
|
|
type_id,
|
|
{
|
|
"id": item.get("structure_id"),
|
|
"origin": "structures",
|
|
"chainage_m": item.get("chainage_m"),
|
|
"start_m": start,
|
|
"end_m": end,
|
|
"count": 1,
|
|
# 원단위 전개와 같은 규칙 — 제원 연장이 있으면 그것, 없으면 시·종점 거리.
|
|
"length_m": stated if stated else span,
|
|
"length_basis": "제원 연장" if stated else ("시·종점" if span else ""),
|
|
"memo": item.get("memo") or "",
|
|
"options": options,
|
|
},
|
|
)
|
|
|
|
for point in pipe_points:
|
|
facility = str(point.get("facility") or "pipe")
|
|
type_id = FACILITY_TYPES.get(facility, facility)
|
|
chainage = float(point.get("chainage_m") or 0.0)
|
|
length = _nearest(lengths, chainage) if type_id == "pipe" else None
|
|
add(
|
|
type_id,
|
|
{
|
|
"id": pipe_row_id(point),
|
|
"origin": "pipe_points",
|
|
"chainage_m": chainage,
|
|
"start_m": point.get("start_m"),
|
|
"end_m": point.get("end_m"),
|
|
"count": 1,
|
|
"length_m": length,
|
|
"length_basis": "B06 횡단 관 연장" if length else "",
|
|
"memo": "",
|
|
"options": dict(point.get("options") or {}),
|
|
},
|
|
)
|
|
|
|
tables = []
|
|
for type_id, definition in types.items():
|
|
rows = rows_by_type.get(type_id)
|
|
if not rows:
|
|
continue
|
|
rows.sort(key=lambda row: float(row.get("chainage_m") or row.get("start_m") or 0.0))
|
|
columns = [
|
|
{
|
|
"key": f.key,
|
|
"label": f.label,
|
|
"unit": f.unit or "",
|
|
"input": f.input,
|
|
"choices": list(f.choices),
|
|
"editable": f.phase == "detail" and f.enabled,
|
|
}
|
|
for f in definition.options
|
|
]
|
|
averages = {}
|
|
for column in columns:
|
|
values = [_number(row["cells"][column["key"]]["value"]) for row in rows]
|
|
present = [v for v in values if v is not None]
|
|
if present and column["input"] == "number":
|
|
averages[column["key"]] = sum(present) / len(present)
|
|
replaced = sum(1 for row in rows if row.get("replaced"))
|
|
if replaced:
|
|
notes.append(
|
|
f"{definition.name} {replaced}줄 — 이 표에서 고친 값을 B05 가 바꿔 "
|
|
"자동값으로 돌아감"
|
|
)
|
|
measured = [row["length_m"] for row in rows if row["length_m"]]
|
|
# 연장이 뜻 있는 표만 빈 연장을 셈 — 반사경·표지판 같은 점 시설은 연장이 없는 것이 정상.
|
|
needs_length = definition.placement == "interval" or type_id == "pipe"
|
|
tables.append(
|
|
{
|
|
"type_id": type_id,
|
|
"name": definition.name,
|
|
"group": definition.group,
|
|
"placement": definition.placement,
|
|
"columns": columns,
|
|
"rows": rows,
|
|
"count": len(rows),
|
|
"length_total_m": sum(measured) if measured else None,
|
|
"length_missing": len(rows) - len(measured) if needs_length else 0,
|
|
"averages": averages,
|
|
}
|
|
)
|
|
return {"tables": tables, "notes": notes}
|
|
|
|
|
|
def pipe_row_id(point: dict[str, Any]) -> str:
|
|
"""계곡 통과 시설 줄 id — 관 지점엔 식별자가 없어 기준점(㎜ 자리)으로 가름."""
|
|
return f"pipe@{float(point.get('chainage_m') or 0.0):.3f}"
|
|
|
|
|
|
def _type_of(target: dict[str, Any], is_pipe: bool) -> str:
|
|
if is_pipe:
|
|
facility = str(target.get("facility") or "pipe")
|
|
return FACILITY_TYPES.get(facility, facility)
|
|
return str(target.get("type_id") or "")
|
|
|
|
|
|
def coerce(field: Any, value: Any, name: str) -> Any:
|
|
"""칸 값 꼴 맞추기 — B05 저장 관문과 같은 규칙(수 칸은 0 이상 수 · 고르기 칸은 보기 안).
|
|
|
|
빈 값(None·"")은 `None` = 그 칸을 지움. 틀리면 `ValueError`.
|
|
"""
|
|
if _blank(value):
|
|
return None
|
|
if field.input == "number":
|
|
number = _number(value)
|
|
if number is None or not math.isfinite(number) or number < 0:
|
|
raise ValueError(f"{name} {field.label}: 0 이상의 수여야 함 — {value!r}")
|
|
return int(number) if number.is_integer() else number
|
|
# 수로 온 보기 값(뒷길이 35 → 35.0)은 보기 글과 같은 꼴로 — 「35.0」은 보기 「35」와 안 맞음.
|
|
whole = isinstance(value, float) and value.is_integer()
|
|
text = str(int(value)) if whole else str(value).strip()
|
|
if field.input == "select" and field.choices and text not in field.choices:
|
|
raise ValueError(f"{name} {field.label}: 보기에 없는 값 — {text}")
|
|
return text
|
|
|
|
|
|
def apply_edits(
|
|
structures: list[dict[str, Any]],
|
|
points: list[dict[str, Any]],
|
|
types: dict[str, Any],
|
|
edits: Iterable[dict[str, Any]],
|
|
marks: dict[str, dict[str, Any]],
|
|
) -> tuple[list[str], dict[str, dict[str, Any]]]:
|
|
"""손 고침을 **정본 사본에 바로** 얹음(제자리 바꿈) — (바뀐 줄 id, 새 손댄 칸 표).
|
|
|
|
고치기 전 값(`was`)은 처음 손댄 때의 정본 값을 지킴 — 그 값으로 되돌리면 표에서 빠져 「자동」.
|
|
놓기 칸·없는 줄·틀린 값은 `LookupError`/`ValueError` — 부르는 쪽이 아무것도 안 씀.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSheet_Edit import EDITABLE_KEYS as SHEET_KEYS
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSheet_Edit import clean_spec
|
|
|
|
rows = {str(item.get("structure_id")): (item, False) for item in structures}
|
|
rows |= {pipe_row_id(point): (point, True) for point in points}
|
|
marks = {row_id: dict(cells) for row_id, cells in marks.items()}
|
|
changed: list[str] = []
|
|
for edit in edits:
|
|
row_id, key = str(edit.get("id")), str(edit.get("key"))
|
|
if row_id not in rows:
|
|
raise LookupError(f"집계표 줄을 못 찾음: {row_id}")
|
|
target, is_pipe = rows[row_id]
|
|
definition = types.get(_type_of(target, is_pipe))
|
|
field = next((f for f in definition.options if f.key == key), None) if definition else None
|
|
if field is None:
|
|
raise ValueError(f"{row_id}: 그 종류에 없는 칸 — {key}")
|
|
if field.phase != "detail":
|
|
raise ValueError(
|
|
f"{definition.name} {field.label}: 자리·길이·높이 같은 놓기 칸은 "
|
|
"시·종점과 한 벌이라 구조물 놓기(B05)에서 고침"
|
|
)
|
|
value = coerce(field, edit.get("value"), definition.name)
|
|
if value is not None and key in SHEET_KEYS:
|
|
# 구조물도 [제원 저장]과 **같은 꼴**로 적음(뒷길이는 정수 등) — 두 길의 꼴이 같아야 함.
|
|
# ⚠ 검사(`coerce`)를 먼저 — `clean_spec` 은 못 읽은 값을 빼 버려 「지우기」로 읽힘.
|
|
value = clean_spec({key: value})[0].get(key, value)
|
|
options = dict(target.get("options") or {})
|
|
old = None if _blank(options.get(key)) else options[key]
|
|
if (value is None and old is None) or (value is not None and _same(old, value)):
|
|
continue
|
|
if value is None:
|
|
options.pop(key, None)
|
|
else:
|
|
options[key] = value
|
|
target["options"] = options
|
|
cells = marks.setdefault(row_id, {})
|
|
mark = cells.get(key)
|
|
# 옛 손 표가 정본과 같을 때만 그 「고치기 전」을 이어 씀 — B05 가 바꾼 뒤면 지금 값이 기준.
|
|
was = mark.get("was") if mark and _same(mark.get("value"), old) else old
|
|
if _same(was, value):
|
|
cells.pop(key, None)
|
|
else:
|
|
cells[key] = {"value": value, "was": was}
|
|
if not cells:
|
|
marks.pop(row_id)
|
|
if row_id not in changed:
|
|
changed.append(row_id)
|
|
return changed, marks
|
|
|
|
|
|
def prune_marks(
|
|
marks: dict[str, dict[str, Any]],
|
|
structures: Iterable[dict[str, Any]],
|
|
points: Iterable[dict[str, Any]],
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""[저장] 때 손댄 칸 표 정리 — 없어진 줄과 **B05 가 이미 바꾼 칸**(알림을 본 뒤)은 뺌."""
|
|
current = {str(item.get("structure_id")): item for item in structures}
|
|
current |= {pipe_row_id(point): point for point in points}
|
|
kept: dict[str, dict[str, Any]] = {}
|
|
for row_id, cells in marks.items():
|
|
if row_id not in current:
|
|
continue
|
|
options = current[row_id].get("options") or {}
|
|
alive = {
|
|
key: mark
|
|
for key, mark in cells.items()
|
|
if _same(None if _blank(options.get(key)) else options[key], mark.get("value"))
|
|
}
|
|
if alive:
|
|
kept[row_id] = alive
|
|
return kept
|
|
|
|
|
|
def pipe_lengths_from_designs(designs: Iterable[dict[str, Any]]) -> dict[float, float]:
|
|
"""저장된 횡단 설계 → `{측점: 관 연장}` — 배수관 물량과 같은 읽기."""
|
|
from B08_Quantity.B08_Quantity_Engine_Pipe import _length_by_chainage
|
|
|
|
return _length_by_chainage(list(designs or []), "pipe_length_m")
|