Merge remote-tracking branch 'origin/sub_laptop_5' into sub_laptop_2

This commit is contained in:
2026-09-29 12:27:01 +09:00
10 changed files with 632 additions and 13 deletions
@@ -269,7 +269,10 @@ async def _resolve(
if context is None:
return JSONResponse(status_code=404, content={"status": "error", "message": reason})
requested = parse_pipe_points((payload or {}).get("points"))
try:
requested = parse_pipe_points((payload or {}).get("points"))
except ValueError as error: # 집계 항목(summary_item) 모양이 틀림
return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
signature = route_signature(context.vertices)
stored = (
load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
@@ -349,6 +352,11 @@ async def put_pipe_points(
_basin_features(context, detail, points),
context.crs,
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from common_util.common_util_storage import resolve_stored_project_path
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
await asyncio.to_thread(safe_sync, resolve_stored_project_path(context.stored_path))
result = _payload(project_id, context, detail, points, saved=True)
result["saved_count"] = saved
return result
@@ -8,6 +8,7 @@
쪽이 앞의 편집을 조용히 지우지 않게.
"""
import asyncio
import logging
from pathlib import Path
from typing import Any
@@ -282,6 +283,10 @@ async def write_structures(
base_revision=payload.base_revision,
max_chainage_m=await _route_length(project_id),
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
await asyncio.to_thread(safe_sync, root)
# 설계에 영향을 주는 변경일 때만 B06 이후를 STALE로 돌린다 — 메모만 고쳐도
# 횡단·수량을 다시 돌리게 만들지 않기 위함이다. 응답 플래그는 실제로 STALE
# 전파가 **성공했을 때만** true(실패를 성공처럼 알리지 않는다).
+10 -1
View File
@@ -13,7 +13,9 @@ import os
from functools import lru_cache
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from common_util.common_util_summary_item import parse_summary_item
# 배치형태: 점형(측점 1개) / 구간형(시~종점) / 부지형(위치+면적).
STRUCTURE_PLACEMENTS = ("point", "interval", "site")
@@ -156,6 +158,13 @@ class StructureInstance(BaseModel):
revision: int = Field(default=0, ge=0)
# 부지형 polygon 예약 — 1차 UI는 측점+면적 속성만 쓴다.
geometry: dict[str, Any] | None = None
# 구조물 집계표 항목(PLAN 21-1) — 없으면 null(옛 파일 호환) · [저장] 뒤 프로젝트 스냅샷
summary_item: dict[str, Any] | None = None
@field_validator("summary_item")
@classmethod
def validate_summary_item(cls, value: Any) -> dict[str, Any] | None:
return parse_summary_item(value)
@model_validator(mode="after")
def validate_placement_fields(self) -> "StructureInstance":
+197
View File
@@ -0,0 +1,197 @@
"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3).
[저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤:
- 구조물 · 관 지점이 가리키는 항목(`summary_item`)마다 원본 층에서 내용을 복사
(`집계.json` SS 줄 + 열 정의 · `도면.json` SD · `면적.json` SA · `재료.json` SM ·
`자재조합.json` UM 줄 · `일위대가.json` UA 줄 + 담은 로직 줄 · `manifest.json`)
- 이미 있는 폴더는 내용을 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌어도 프로젝트
값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로
- 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제)
원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가는 M01 시스템 것.
"""
from __future__ import annotations
import hashlib
import json
import logging
import shutil
from datetime import datetime
from pathlib import Path
from typing import Any
from B05_Profile.B05_Profile_Structures_Repository import load_structures
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_json import atomic_write_json
from common_util.common_util_summary_item import snapshot_folder_name
from M02_MasterTemplete import M02_Template_Layers as layers
logger = logging.getLogger(__name__)
ITEMS_DIRNAME = "items"
MANIFEST_NAME = "manifest.json"
TABLE_NAME = "구조물집계표"
UM_FILE = "자재조합.json"
UA_FILE = "일위대가조합.json" # M01 시스템(`resources/master_data/`)
#: 열 id → 스냅샷 파일(구조물 도면 · 면적산출 · 재료산출)
PART_FILES = (("structure", "도면.json"), ("area", "면적.json"), ("material", "재료.json"))
def items_dir(project_root: str | Path) -> Path:
return layers.project_dir(project_root) / ITEMS_DIRNAME
def layer_root(item: dict[str, Any]) -> Path:
"""항목 원본 층 뿌리 — 시스템 · 회사 · 개인(21-2 층 폴더 함수가 오면 그리로)."""
if item["layer"] == "system":
return layers.system_dir()
if item["layer"] == "company":
return layers.company_dir(item["owner"])
company, user = str(item["owner"]).split("/")
return layers.personal_dir(company, user)
def _read(path: Path) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
def _row(document: Any, key: str) -> dict[str, Any] | None:
rows = document.get("줄") if isinstance(document, dict) else None
for row in rows or []:
if isinstance(row, dict) and row.get("키") == key:
return row
return None
def _labor_rows(ua_key: str) -> dict[str, Any] | None:
"""UA 줄 + 담은 로직 줄(M01 시스템 · 없는 로직은 빼고 키만 `빠진로직`)."""
from M01_MasterData import M01_MasterData_Store as m01
ua = _row(_read(Path(m01.FOLDER) / UA_FILE), ua_key)
if ua is None:
return None
wanted = [str(entry.get("로직")) for entry in ua.get("담은로직") or [] if entry.get("로직")]
found: dict[str, dict[str, Any]] = {}
raw: dict[str, Any] = {} # M01 적재본은 수가 Decimal — 줄은 원본 파일 JSON 그대로 담음
for key in wanted:
try:
file = m01.logic(key)["file"]
except m01.StoreError:
continue
if file not in raw:
raw[file] = _read(Path(m01.FOLDER) / file)
row = _row(raw[file], key)
if row is not None:
found[key] = row
return {
"줄": ua,
"로직": [found[key] for key in wanted if key in found],
"빠진로직": [key for key in wanted if key not in found],
}
def build_snapshot(item: dict[str, Any]) -> dict[str, Any]:
"""원본 층에서 항목 내용을 모음 — `{파일 이름: 내용}` · SS 줄이 없으면 LookupError."""
root = layer_root(item)
row = _row(_read(root / layers.COMBO_FILE), item["key"])
if row is None:
raise LookupError(f"집계 항목 {item['key']} 이 {item['layer']} 층에 없음")
column_id = str(row.get("열") or "")
table = _read(root / "table" / f"{TABLE_NAME}.json") or {}
column = next((c for c in table.get("열") or [] if c.get("id") == column_id), None)
files: dict[str, Any] = {"집계.json": {"줄": row, "열": column}}
for kind, name in PART_FILES:
part = _read(root / kind / f"{column_id}.json") if column_id else None
if part is not None:
files[name] = part
if row.get("자재조합"):
um = _row(_read(root / UM_FILE), str(row["자재조합"]))
if um is not None:
files["자재조합.json"] = um
if row.get("일위대가"):
labor = _labor_rows(str(row["일위대가"]))
if labor is not None:
files["일위대가.json"] = labor
return files
def assignments(project_root: str | Path) -> dict[str, tuple[dict[str, Any], list[str]]]:
"""`{폴더 이름: (항목, 쓰는 구조물 id)}` — 구조물 정본 + 관 지점 정본(id = `pipe@측점`)."""
found: dict[str, tuple[dict[str, Any], list[str]]] = {}
def add(item: dict[str, Any] | None, owner_id: str) -> None:
if item:
found.setdefault(snapshot_folder_name(item), (item, []))[1].append(owner_id)
_, structures = load_structures(str(project_root))
for index, structure in enumerate(structures):
add(structure.summary_item, structure.structure_id or f"#{index}")
for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
add(point.summary_item, f"pipe@{round(float(point.chainage_m), 3)}")
return found
def _version(files: dict[str, Any]) -> str:
raw = json.dumps(files, ensure_ascii=False, sort_keys=True).encode("utf-8")
return hashlib.sha256(raw).hexdigest()[:16]
def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
"""지정 항목 스냅샷을 맞춤 — `{만듦, 지움, 못만듦}` 폴더 이름."""
base = items_dir(project_root)
wanted = assignments(project_root)
made: list[str] = []
failed: list[str] = []
for folder, (item, users) in sorted(wanted.items()):
target = base / folder
manifest = _read(target / MANIFEST_NAME)
if isinstance(manifest, dict):
if manifest.get("구조물") != users:
atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": users})
continue
try:
files = build_snapshot(item)
except (LookupError, ValueError) as error:
logger.warning("집계 항목 스냅샷을 못 만듦 (%s): %s", folder, error)
failed.append(folder)
continue
target.mkdir(parents=True, exist_ok=True)
for name, content in files.items():
atomic_write_json(target / name, content)
# manifest 는 마지막 — 있으면 다 쓴 폴더(도중에 끊기면 다음 저장이 다시 씀)
atomic_write_json(
target / MANIFEST_NAME,
{
"원본": item,
"판": _version(files),
"적용일": datetime.now().isoformat(timespec="seconds"),
"구조물": users,
},
)
made.append(folder)
removed: list[str] = []
if base.is_dir():
for child in sorted(base.iterdir()):
if child.is_dir() and child.name not in wanted:
shutil.rmtree(child)
removed.append(child.name)
return {"만듦": made, "지움": removed, "못만듦": failed}
def safe_sync(project_root: str | Path) -> None:
"""저장 뒤 부름 — 실패해도 저장은 그대로(스냅샷은 다음 저장이 다시 맞춤)."""
try:
result = sync_snapshots(project_root)
if any(result.values()):
logger.info("집계 항목 스냅샷: %s", result)
except Exception:
logger.exception("집계 항목 스냅샷 맞춤 실패 (저장은 유지): %s", project_root)
def read_snapshot(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any] | None:
"""프로젝트 스냅샷의 `집계.json`(`{줄, 열}`) — 없으면 None."""
data = _read(items_dir(project_root) / snapshot_folder_name(item) / "집계.json")
return data if isinstance(data, dict) else None
+83 -6
View File
@@ -106,6 +106,7 @@ def collect_items(
"start_m": item.start_m,
"end_m": item.end_m,
"options": _with_defaults(definition, dict(item.options or {})),
"summary_item": item.summary_item,
}
)
for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
@@ -120,6 +121,7 @@ def collect_items(
"end_m": None,
"options": _with_defaults(types.get(type_id), dict(point.options or {})),
"관연장": _nearest(lengths, chainage) if type_id == "pipe" else None,
"summary_item": point.summary_item,
}
)
return items
@@ -186,6 +188,27 @@ def _sources(binding: dict[str, Any]) -> list[dict[str, Any]]:
return [main, *(binding.get("더함") or [])]
def _home(column: dict[str, Any]) -> str:
"""펼친 열이면 마스터 열의 묶음 — 지정 항목이 고른 열(묶음)과 견줌."""
return _group(_frame(column).get("원열") or column)
def _assigned_match(source: dict[str, Any], item: dict[str, Any], column: dict[str, Any]) -> bool:
"""지정 항목 모드 — 구조물이 고른 열(묶음)에만 · 종류 · 조건 · 옵션으로 자동 맞추지 않음.
값을 읽을 출처 = 바인딩 출처 가운데 그 구조물 종류 것(없으면 주 출처).
"""
if item.get("열묶음") != _home(column):
return False
sources = _sources(_frame(column))
typed = [entry for entry in sources if entry.get("종류") == item["type_id"]]
return source == (typed[0] if typed else sources[0])
def _auto_match(source: dict[str, Any], item: dict[str, Any], _column: dict[str, Any]) -> bool:
return _matches(source, item)
def _expansion(source: dict[str, Any], item: dict[str, Any]) -> tuple[str, ...]:
values = []
for key in source.get("펼침") or []:
@@ -313,6 +336,7 @@ def _expand_columns(
items: list[dict[str, Any]],
variables: dict[str, Any],
choices: dict[str, list[str]],
match: Any = _auto_match,
) -> tuple[list[dict[str, Any]], dict[str, list[tuple[str, str, dict[str, Any]]]]]:
"""펼침 열 → 값마다 열 · 묶음은 값마다 한데(관 Φ800 관매설 · 커플링 → Φ1000 …).
@@ -333,7 +357,7 @@ def _expand_columns(
found = groups.setdefault(_group(column), set())
for source in _sources(binding):
# 값이 빈 구조물도 열은 세움 — 놓였는데 수량이 빈칸인 것이 보여야 함
found.update(_expansion(source, item) for item in items if _matches(source, item))
found.update(_expansion(source, item) for item in items if match(source, item, column))
out: list[dict[str, Any]] = []
made: dict[str, list[tuple[str, str, dict[str, Any]]]] = {}
@@ -394,14 +418,19 @@ def fill_document(
document: dict[str, Any],
items: list[dict[str, Any]],
types: dict[str, StructureType] | None = None,
assigned: bool = False,
) -> dict[str, Any]:
"""표 문서 + 설계 구조물 목록 → 채운 표 문서(새 사본)."""
"""표 문서 + 설계 구조물 목록 → 채운 표 문서(새 사본).
`assigned` = 지정 항목 모드(`fill_table`) — 구조물마다 고른 열(`열묶음`)에만 넣음.
"""
match = _assigned_match if assigned else _auto_match
types = types or structure_type_map()
doc = copy.deepcopy(document)
variables = dict(doc.get("변수") or {})
interval = _number(variables.get("측점간격_m")) or 20.0
rule_columns = _unexpand(doc.get("열") or [])
columns, made = _expand_columns(rule_columns, items, variables, _choices(types))
columns, made = _expand_columns(rule_columns, items, variables, _choices(types), match)
notes: list[str] = []
# 줄 자리 — 관은 한 줄에 하나(둘째 관부터 줄을 나눔)
@@ -427,7 +456,7 @@ def fill_document(
if source.get("값") in (None, "식"):
continue
for item, slot in zip(items, item_slot, strict=True):
if not _matches(source, item):
if not match(source, item, column):
continue
if expanded is not None and list(_expansion(source, item)) != expanded:
continue
@@ -558,14 +587,62 @@ def is_fillable(document: Any) -> bool:
)
def assign_items(
project_root: str | Path, document: dict[str, Any], items: list[dict[str, Any]]
) -> tuple[dict[str, Any], list[dict[str, Any]], list[str]]:
"""지정 항목 모드 준비 — 구조물마다 스냅샷 열 정의로 고른 열을 정함(PLAN 21-0 · 21-3).
- 미지정 · 스냅샷 없음 = 표에 안 넣음(자동 맞춤 없음) · 알림 한 줄씩
- 시스템 층 항목 = 같은 id 열 자리에 스냅샷 열 정의(원본이 바뀌어도 지정한 때 값)
- 회사 · 개인 층 항목 = `열id@스냅샷폴더` 새 열(시스템 열과 id · 묶음이 안 겹치게)
돌려주는 것 = (열을 맞춘 문서, `열` · `열묶음` 을 단 구조물, 알림).
"""
from common_util.common_util_summary_item import snapshot_folder_name
from M02_MasterTemplete.M02_Item_Snapshot import read_snapshot
doc = copy.deepcopy(document)
rules = _unexpand(doc.get("열") or [])
out: list[dict[str, Any]] = []
unassigned = missing = 0
for item in items:
chosen = item.get("summary_item")
column = (read_snapshot(project_root, chosen) or {}).get("열") if chosen else None
if not chosen or not isinstance(column, dict) or not column.get("id"):
unassigned += not chosen
missing += bool(chosen)
continue
new = copy.deepcopy(column)
if chosen["layer"] != "system":
new["id"] = f"{column['id']}@{snapshot_folder_name(chosen)}"
if isinstance(new.get("바인딩"), dict) and new["바인딩"].get("묶음"):
new["바인딩"]["묶음"] = new["id"]
at = next((i for i, c in enumerate(rules) if c.get("id") == new["id"]), None)
if at is None:
rules.append(new)
else:
rules[at] = new
out.append({**item, "열": new["id"], "열묶음": _group(new)})
doc["열"] = rules
notes = []
if unassigned:
notes.append(f"집계 항목 미지정 {unassigned}건 — 표에 안 들어감")
if missing:
notes.append(f"집계 항목 복사본 없음 {missing}건 — [저장] 뒤 채워짐")
return doc, out, notes
def fill_table(
project_root: str | Path,
document: dict[str, Any],
pipe_lengths: dict[float, float] | None = None,
) -> dict[str, Any]:
"""프로젝트 설계로 표 문서를 채움 — 저장 안 함."""
"""프로젝트 설계로 표 문서를 채움 — 저장 안 함 · 지정 항목 모드(미지정은 빈칸)."""
types = structure_type_map()
return fill_document(document, collect_items(project_root, pipe_lengths, types), types)
items = collect_items(project_root, pipe_lengths, types)
doc, chosen, notes = assign_items(project_root, document, items)
filled = fill_document(doc, chosen, types, assigned=True)
filled["알림"] = [*notes, *filled.get("알림", [])]
return filled
def recalc(document: dict[str, Any]) -> dict[str, Any] | None:
@@ -35,6 +35,7 @@ from shapely.geometry import LineString, Point
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex
from common_util.common_util_summary_item import parse_summary_item
from config.config_system import (
DRAINAGE_CACHE_DIRNAME,
DRAINAGE_DETAIL_FILENAME,
@@ -89,6 +90,8 @@ class PipePoint:
# 관이 실제로 놓인 자리(사업지 CRS, m). 노선이 바뀌어도 이 자리는 그대로다.
x: float | None = None
y: float | None = None
# 구조물 집계표 항목(PLAN 21-1 · options 밖) — 없으면 None · 모양은 `common_util_summary_item`
summary_item: dict[str, Any] | None = None
def as_dict(self) -> dict[str, Any]:
# 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
@@ -110,6 +113,8 @@ class PipePoint:
if self.x is not None and self.y is not None:
payload["x"] = round(float(self.x), 3)
payload["y"] = round(float(self.y), 3)
if self.summary_item:
payload["summary_item"] = self.summary_item
return payload
@@ -349,6 +354,7 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
if isinstance(options, dict) and options
else None
),
summary_item=parse_summary_item(item.get("summary_item")),
)
)
points.sort(key=lambda point: point.chainage_m)
@@ -399,6 +405,7 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint])
point.start_m = ref.start_m
point.end_m = ref.end_m
point.options = ref.options
point.summary_item = ref.summary_item
return base
+61
View File
@@ -0,0 +1,61 @@
"""구조물 집계표 항목 참조 `summary_item` — 모양 검사 · 스냅샷 폴더 이름 (PLAN 21-1).
summary_item = {layer: "system"|"company"|"personal",
owner: null|"{회사}"|"{회사}/{사용자}", key: "SS000004", name: "…"}
구조물 기록(`structures.json` `StructureInstance`) · 관 지점 기록(`pipe_points.json` 점) 두 곳이
같은 칸을 씀 — 검사는 여기 한 벌. 없으면 null(옛 파일 호환).
"""
from __future__ import annotations
import re
from typing import Any
SUMMARY_LAYERS = ("system", "company", "personal")
_KEY = re.compile(r"^SS\d{6}$")
_OWNER = {
"system": None,
"company": re.compile(r"^\d+$"),
"personal": re.compile(r"^\d+/\d+$"),
}
_NAME_MAX = 200
def parse_summary_item(value: Any) -> dict[str, Any] | None:
"""모양이 맞으면 `{layer, owner, key, name}` 새 dict · 빈 값은 None · 틀리면 ValueError."""
if value is None:
return None
if not isinstance(value, dict):
raise ValueError("집계 항목(summary_item)은 객체여야 합니다.")
extra = set(value) - {"layer", "owner", "key", "name"}
if extra:
raise ValueError(f"집계 항목에 모르는 칸이 있습니다: {', '.join(sorted(extra))}")
layer = value.get("layer")
if layer not in SUMMARY_LAYERS:
raise ValueError("집계 항목 층은 system · company · personal 중 하나입니다.")
owner = value.get("owner")
rule = _OWNER[layer]
if rule is None:
if owner is not None:
raise ValueError("시스템 층 집계 항목의 owner 는 null 이어야 합니다.")
elif not isinstance(owner, str) or not rule.match(owner):
shape = "{회사}" if layer == "company" else "{회사}/{사용자}"
raise ValueError(f"{layer} 층 집계 항목의 owner 는 「{shape}」 숫자 모양입니다.")
key = value.get("key")
if not isinstance(key, str) or not _KEY.match(key):
raise ValueError("집계 항목 key 는 SS 와 숫자 6자리입니다(예: SS000004).")
name = value.get("name")
if not isinstance(name, str) or not name.strip() or len(name) > _NAME_MAX:
raise ValueError(f"집계 항목 name 은 1~{_NAME_MAX}자 글입니다.")
return {"layer": layer, "owner": owner, "key": key, "name": name}
def snapshot_folder_name(item: dict[str, Any]) -> str:
"""`{layer}_{owner 슬러그}_{key}` — owner 없음 = `none` · `/` = `-`.
예: `system_none_SS000004` · `company_3_SS000001` · `personal_3-12_SS000002`.
"""
owner = item.get("owner")
slug = str(owner).replace("/", "-") if owner else "none"
return f"{item['layer']}_{slug}_{item['key']}"
+9 -3
View File
@@ -71,8 +71,14 @@ def project(tmp_path: Path) -> Path:
return root
def _auto(project: Path, document: dict[str, Any], lengths: dict[float, float]) -> dict[str, Any]:
"""바인딩 자동 맞춤(`fill_document` 기본) — 프로젝트 `fill_table` 은 지정 항목 모드(PLAN 21-3)라
펼침 · 묶음 · 손 값 규칙은 이 길로 봄."""
return fill.fill_document(document, fill.collect_items(project, lengths))
def _filled(project: Path, document: dict[str, Any] | None = None) -> dict[str, Any]:
return fill.fill_table(project, document or _master(), {85.0: 14.0})
return _auto(project, document or _master(), {85.0: 14.0})
def _col(doc: dict[str, Any], col_id: str) -> dict[str, Any]:
@@ -119,7 +125,7 @@ def test_계산_열은_설계값_딱지_없음(project: Path) -> None:
assert "바인딩" not in column and column["펼침틀"]["원열"]["id"] == col_id.split("|")[0]
assert _col(doc, "pp_cp|파형강관|1000")["머리"] == ["관공", "파형강관 Φ1000", "커플링밴드"]
# 다시 저장 · 채워도 같은 열(계산 열도 원열로 되돌림)
again = fill.fill_table(project, fill.strip_design(doc), {85.0: 14.0})
again = _auto(project, fill.strip_design(doc), {85.0: 14.0})
assert [c["id"] for c in again["열"]] == [c["id"] for c in doc["열"]]
@@ -152,7 +158,7 @@ def test_측점_줄_같은_측점은_한_줄_관은_나눔(project: Path) -> Non
assert [row["값"].get("no") for row in doc["줄"][1:]] == [str(i) for i in range(1, 8)]
assert "식" not in rows["s200.00~230.00"] # 줄눈 간격이 달라도 열 식이 없으면 줄 식 없음
assert doc["알림"] == []
bare = fill.fill_table(project, _master(), {})
bare = _auto(project, _master(), {})
assert bare["알림"] == ["관 연장 없음 2곳 — B06 횡단 설계 전이면 빈칸"]
assert not any("pp_len" in key for row in bare["줄"] for key in row["값"])
+3 -2
View File
@@ -207,8 +207,9 @@ def test_채운_표는_작업본을_채워_돌려주고_저장_안_함(
)
before = layers.version_of(root / "templates/table/구조물집계표.json")
got = world["client"].get(f"/api/m02/projects/{P1}/tables/구조물집계표/filled").json()
row = next(r for r in got["문서"]["줄"] if r["id"] == "s40.00")
assert row["값"]["sta"] == "NO.2" and row["값"]["pp_len|파형강관|1000"] == 12
# 프로젝트 채움 = 지정 항목 모드(PLAN 21-3) — 집계 항목 미지정 관은 표에 안 들어감
assert not any(r["id"] == "s40.00" for r in got["문서"]["줄"])
assert "집계 항목 미지정 1건 — 표에 안 들어감" in got["문서"]["알림"]
assert got["결과"] == {"계산": {}} and got["판"] == before
assert layers.version_of(root / "templates/table/구조물집계표.json") == before
@@ -0,0 +1,248 @@
"""구조물 집계표 항목 지정 · 저장 스냅샷 · 채움 (PLAN 21-3 · 임시 저장소).
- `summary_item` 칸(구조물 · 관 지점) 모양 검사 · 옛 파일(칸 없음) 호환
- 저장 뒤 스냅샷: 지정 → 폴더 생김 · 원본(마스터) 바꿔도 프로젝트 값 그대로 · 지정 바꿈 → 옛 폴더
지움 · 해제 → 지움 · 회사 층 자리
- 채움(`fill_table`) = 지정 항목 모드 — 고른 열에만 · 미지정은 안 셈 · 스냅샷 열 정의 사용
- [초기화]는 `templates/` 를 안 건드림 · 프로젝트 복제 도구는 폴더를 통째 복사
"""
from __future__ import annotations
import json
import shutil
from pathlib import Path
from typing import Any
import pytest
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance
from common_util import common_util_initial_snapshot as initial
from common_util.common_util_drainage_pipes import PipePoint, parse_pipe_points
from common_util.common_util_summary_item import parse_summary_item
from config import config_system
from M02_MasterTemplete import M02_Item_Snapshot as snap
from M02_MasterTemplete import M02_Table_Fill as fill
from M02_MasterTemplete import M02_Template_Layers as layers
REAL = config_system.PROJECT_ROOT / "resources/master_template"
MS = {"layer": "system", "owner": None, "key": "SS000022", "name": "돌쌓기"}
RV = {"layer": "system", "owner": None, "key": "SS000001", "name": "돌기슭막이 메쌓기 1.2m"}
PIPE = {"layer": "system", "owner": None, "key": "SS000011", "name": "관공 관매설"}
def _copy_layer(target: Path) -> None:
target.mkdir(parents=True)
for name in (layers.COMBO_FILE, "자재조합.json"):
shutil.copy2(REAL / name, target / name)
for kind in ("table", "structure", "area", "material"):
shutil.copytree(REAL / kind, target / kind)
@pytest.fixture
def world(tmp_path: Path, monkeypatch) -> dict[str, Any]:
system = tmp_path / "master"
_copy_layer(system)
monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
storage = tmp_path / "storage"
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
root = storage / "3" / "12" / "proj"
(root / "B05_Profile/route").mkdir(parents=True)
(root / "B04_PreProcess/drainage/edits").mkdir(parents=True)
return {"system": system, "storage": storage, "root": root}
def _save(root: Path, structures: list[dict], points: list[dict]) -> dict[str, list[str]]:
"""화면 [저장] 두 곳(구조물 · 관 지점)이 쓴 뒤 서버가 맞추는 것과 같게."""
for index, item in enumerate(structures):
item.setdefault("structure_id", f"st{index}")
item.setdefault("placement", "interval" if "start_m" in item else "point")
(root / "B05_Profile/route/structures.json").write_text(
json.dumps({"revision": 1, "structures": structures}, ensure_ascii=False), encoding="utf-8"
)
(root / "B04_PreProcess/drainage/edits/pipe_points.json").write_text(
json.dumps({"points": points}, ensure_ascii=False), encoding="utf-8"
)
return snap.sync_snapshots(root)
def _dry(summary: dict | None) -> dict[str, Any]:
wall = {"type_id": "masonry_dry", "chainage_m": 85, "start_m": 80, "end_m": 90}
return {**wall, "options": {"height_m": 2.0, "length_m": 10}, "summary_item": summary}
def _wet() -> dict[str, Any]:
wall = {"type_id": "masonry_wet", "chainage_m": 120, "start_m": 120, "end_m": 130}
return {**wall, "options": {"height_m": 2.5, "length_m": 10}}
def _pipe(summary: dict | None) -> dict[str, Any]:
options = {"pipe_kind": "파형강관", "pipe_diameter_mm": "1000"}
return {"chainage_m": 40.0, "source": "user", "options": options, "summary_item": summary}
def _items(root: Path) -> Path:
return root / "templates" / "items"
# ── 칸 모양 ──────────────────────────────────────────
@pytest.mark.parametrize(
"bad",
[
{"layer": "project", "owner": None, "key": "SS000001", "name": "x"},
{"layer": "system", "owner": "3", "key": "SS000001", "name": "x"},
{"layer": "company", "owner": "3/12", "key": "SS000001", "name": "x"},
{"layer": "personal", "owner": "3", "key": "SS000001", "name": "x"},
{"layer": "system", "owner": None, "key": "UA000001", "name": "x"},
{"layer": "system", "owner": None, "key": "SS000001", "name": ""},
{"layer": "system", "owner": None, "key": "SS000001", "name": "x", "more": 1},
"SS000001",
],
)
def test_모양이_틀린_항목은_거절(bad):
with pytest.raises(ValueError):
parse_summary_item(bad)
with pytest.raises(ValueError):
StructureInstance(type_id="x", placement="point", chainage_m=1, summary_item=bad)
with pytest.raises(ValueError):
parse_pipe_points([{"chainage_m": 1.0, "summary_item": bad}])
def test_옛_파일은_칸_없음_그대로_읽고_씀():
item = StructureInstance(type_id="x", placement="point", chainage_m=1)
assert item.summary_item is None
point = parse_pipe_points([{"chainage_m": 10.0, "source": "user"}])[0]
assert point.summary_item is None and "summary_item" not in point.as_dict()
owned = {"layer": "personal", "owner": "3/12", "key": "SS000002", "name": "내 옹벽"}
again = parse_pipe_points([PipePoint(chainage_m=5.0, summary_item=owned).as_dict()])[0]
assert again.summary_item == owned
# ── 스냅샷 ───────────────────────────────────────────
def test_지정하면_폴더가_생기고_내용이_원본과_같음(world):
root = world["root"]
result = _save(root, [_dry(MS), _wet()], [_pipe(PIPE)])
assert result == {
"만듦": ["system_none_SS000011", "system_none_SS000022"],
"지움": [],
"못만듦": [],
}
folder = _items(root) / "system_none_SS000022"
got = {path.name for path in folder.iterdir()}
assert {"집계.json", "도면.json", "면적.json", "재료.json", "manifest.json"} <= got
total = json.loads((folder / "집계.json").read_text(encoding="utf-8"))
assert total["줄"]["키"] == "SS000022" and total["열"]["id"] == "ms"
drawing = json.loads((world["system"] / "structure/ms.json").read_text(encoding="utf-8"))
assert json.loads((folder / "도면.json").read_text(encoding="utf-8")) == drawing
manifest = json.loads((folder / "manifest.json").read_text(encoding="utf-8"))
assert manifest["원본"] == MS and manifest["구조물"] == ["st0"] and len(manifest["판"]) == 16
pipe = json.loads(
(_items(root) / "system_none_SS000011/manifest.json").read_text(encoding="utf-8")
)
assert pipe["구조물"] == ["pipe@40.0"]
# 일위대가 UA 줄 + 담은 로직 줄(M01 시스템)
labor = json.loads(
(_items(root) / "system_none_SS000011/일위대가.json").read_text(encoding="utf-8")
)
assert labor["줄"]["키"] == "UA000036"
assert [row["키"] for row in labor["로직"]] + labor["빠진로직"] == [
entry["로직"] for entry in labor["줄"]["담은로직"]
]
def test_원본을_바꿔도_프로젝트_값은_그대로(world):
root, system = world["root"], world["system"]
_save(root, [_dry(MS)], [])
before = (_items(root) / "system_none_SS000022/집계.json").read_text(encoding="utf-8")
table = json.loads((system / "table/구조물집계표.json").read_text(encoding="utf-8"))
for column in table["열"]:
if column["id"] == "ms":
column["바인딩"]["머리틀"] = ["바뀐 돌쌓기", "{0}", "{1}"]
(system / "table/구조물집계표.json").write_text(
json.dumps(table, ensure_ascii=False), encoding="utf-8"
)
assert _save(root, [_dry(MS)], [])["만듦"] == [] # 다시 저장해도 안 다시 씀
after = (_items(root) / "system_none_SS000022/집계.json").read_text(encoding="utf-8")
assert after == before
doc = fill.fill_table(root, table)
heads = [c["머리"][0] for c in doc["열"] if c["id"].startswith("ms|")]
assert heads == ["돌쌓기"] # 채움도 스냅샷 열 정의
def test_지정_바꿈_해제_삭제면_옛_폴더를_지움(world):
root = world["root"]
_save(root, [_dry(MS), _dry(MS)], [])
manifest = _items(root) / "system_none_SS000022/manifest.json"
assert json.loads(manifest.read_text(encoding="utf-8"))["구조물"] == ["st0", "st1"]
# 한 구조물만 바꿈 — 같은 항목을 쓰는 구조물이 남으면 폴더 유지 · 목록만 줄어듦
result = _save(root, [_dry(RV), _dry(MS)], [])
assert result["만듦"] == ["system_none_SS000001"] and result["지움"] == []
assert json.loads(manifest.read_text(encoding="utf-8"))["구조물"] == ["st1"]
# 해제 · 삭제
assert _save(root, [_dry(None)], [])["지움"] == ["system_none_SS000001", "system_none_SS000022"]
assert list(_items(root).iterdir()) == []
def test_회사_층_항목은_회사_폴더에서_복사(world):
root = world["root"]
company = world["storage"] / "3" / "templates"
_copy_layer(company)
combo = json.loads((company / layers.COMBO_FILE).read_text(encoding="utf-8"))
for row in combo["줄"]:
if row["키"] == "SS000022":
row["분류3"] = "회사 것"
(company / layers.COMBO_FILE).write_text(json.dumps(combo, ensure_ascii=False), "utf-8")
owned = {**MS, "layer": "company", "owner": "3"}
assert _save(root, [_dry(owned)], [])["만듦"] == ["company_3_SS000022"]
total = json.loads((_items(root) / "company_3_SS000022/집계.json").read_text("utf-8"))
assert total["줄"]["분류3"] == "회사 것"
# 채움 — 회사 항목은 시스템 열과 안 겹치는 새 열
doc = fill.fill_table(
root, json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8"))
)
ids = [c["id"] for c in doc["열"] if "@" in c["id"]]
assert ids == ["ms@company_3_SS000022|메쌓기|2.0"]
def test_원본에_없는_항목은_못만듦으로_남김(world):
gone = {**MS, "key": "SS999999"}
assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["system_none_SS999999"]
# ── 채움 ─────────────────────────────────────────────
def test_채움은_지정한_열에만_미지정은_안_셈(world):
root = world["root"]
_save(root, [_dry(MS), _wet()], [_pipe(PIPE), {**_pipe(None), "chainage_m": 60.0}])
table = json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8"))
doc = fill.fill_table(root, table, {40.0: 12.0, 60.0: 9.0})
values = {row["id"]: row["값"] for row in doc["줄"] if not row.get("고정")}
assert set(values) == {"s80.00~90.00", "s40.00"} # 미지정 옹벽 · 관은 줄도 없음
assert values["s80.00~90.00"]["ms|메쌓기|2.0"] == 10 # 메쌓기는 더함 출처(종류)로
assert values["s40.00"]["pp_len|파형강관|1000"] == 12
assert not any("찰쌓기" in c["id"] for c in doc["열"]) # 자동 맞춤 없음
assert "집계 항목 미지정 2건 — 표에 안 들어감" in doc["알림"]
def test_스냅샷이_없으면_알림(world):
root = world["root"]
_save(root, [_dry(MS)], [])
shutil.rmtree(_items(root))
doc = fill.fill_table(
root, json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8"))
)
assert "집계 항목 복사본 없음 1건 — [저장] 뒤 채워짐" in doc["알림"]
# ── 초기화 · 복제 ────────────────────────────────────
def test_초기화는_templates_를_안_건드리고_복제는_통째로():
assert not any(tree.startswith("templates") for tree in initial._FILE_TREES)
tool = (config_system.PROJECT_ROOT / "db_management/tools_clone_project.py").read_text("utf-8")
assert '"robocopy",' in tool and '"/E",' in tool # 프로젝트 폴더 전체(templates/items 포함)