eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
# Conflicts:
#	B04_PreProcess/B04_PreProcess_Router_Basins.py
#	B08_Quantity/B08_Quantity_Router_Summary.py
This commit is contained in:
eomsangdon
2026-10-09 15:54:20 +09:00
parent 5305435095 8d4f4e8718
commit ce33e56a00
25 changed files with 762 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
B04_PreProcess/B04_PreProcess_Api_Fetch.ts
+23 -4
View File
@@ -17,6 +17,7 @@ import {
API_SURFACE_BUILD_TIMEOUT_MS,
API_TIMEOUT_MS,
} from "@config/config_frontend";
import { showToast } from "@ui/ui_template_elements";
import type { GroupItems, SummaryItem } from "../common_util/common_util_summary_item";
/** 지표면 분석 실행 요청 (SurfaceAnalyzeRequest) */
@@ -557,6 +558,22 @@ export interface DetailBasinResponse {
arrow_spacing_m: number;
/** 유역 안쪽 상류 세류망 — 하이라이트 토글용. */
upstream_lonlat: Array<Array<[number, number]>>;
/** [저장] 때 못 만든 구조물 조합 사본 폴더(142-2 · `*` = 통째 실패). */
snapshot_failed?: string[];
}
let lastNote = { text: "", at: 0 };
/** 못 만든 구조물 조합 사본(저장 응답 `snapshot_failed` · 142-2) — [저장] 알림 한 줄 ·
* 관 · 구조물 저장이 같은 맞춤을 두 번 알리면 한 번만. */
export function noteSnapshotFailed(failed: readonly string[] | undefined): void {
if (!failed?.length) return;
const text = failed.includes("*")
? "구조물 조합 사본 맞춤에 실패했습니다 — 다시 [저장]해 주세요."
: `구조물 조합 사본 ${failed.length}곳을 못 만들었습니다(원본 조합 없음) — 조합을 다시 골라 [저장]해 주세요.`;
const now = Date.now();
if (text === lastNote.text && now - lastNote.at < 5000) return;
lastNote = { text, at: now };
showToast(text, "error");
}
/** 저장된 관 매설 지점과 그 세부유역. 저장분이 없으면 백엔드가 자동 생성해 돌려준다. */
@@ -593,8 +610,10 @@ export async function saveDetailPipePoints(
projectId: string,
points: DetailPipeInput[],
): Promise<DetailBasinResponse> {
return requestJson<DetailBasinResponse>(`/projects/${projectId}/drainage/pipe-points`, {
method: "PUT",
body: JSON.stringify({ points }),
});
const saved = await requestJson<DetailBasinResponse>(
`/projects/${projectId}/drainage/pipe-points`,
{ method: "PUT", body: JSON.stringify({ points }) },
);
noteSnapshotFailed(saved.snapshot_failed);
return saved;
}
B04_PreProcess/B04_PreProcess_Router_Basins.py
+10 -4
View File
@@ -20,12 +20,13 @@ import math
from typing import Any
from uuid import UUID
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m
from common_util.common_util_auth import get_optional_session
from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
from common_util.common_util_drainage_detail import DrainageDetail, build_detail
from common_util.common_util_drainage_pipes import (
@@ -422,7 +423,9 @@ async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
@router.put("/{project_id}/drainage/pipe-points", response_model=None)
async def put_pipe_points(
project_id: UUID, payload: dict[str, Any] | None = None
project_id: UUID,
payload: dict[str, Any] | None = None,
session: dict[str, Any] | None = Depends(get_optional_session), # 가져온이(142-2)
) -> dict[str, Any] | JSONResponse:
"""관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
resolved = await _resolve(project_id, payload, use_stored=True)
@@ -443,9 +446,12 @@ async def put_pipe_points(
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from common_util.common_util_storage import resolve_stored_project_path
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
await asyncio.to_thread(safe_sync, resolve_stored_project_path(context.stored_path))
failed = await asyncio.to_thread(
safe_sync, resolve_stored_project_path(context.stored_path), importer(session)
)
result = _payload(project_id, context, detail, points, saved=True)
result["saved_count"] = saved
result["snapshot_failed"] = failed # 못 만든 구조물 조합 복사본(142-2) — [저장] 알림
return result
B05_Profile/B05_Profile_Api_Structures.ts
+9 -4
View File
@@ -12,6 +12,7 @@
* ========================================================================== */
import { readState, writeState } from "../A00_Common/b_page_state";
import { noteSnapshotFailed } from "../B04_PreProcess/B04_PreProcess_Api_Fetch";
import type { GroupItems, SummaryItem } from "../common_util/common_util_summary_item";
import { API_BASE_URL, API_TIMEOUT_MS } from "@config/config_frontend";
import {
@@ -150,6 +151,8 @@ export interface StructureSaveResponse {
needs_downstream_invalidation: boolean;
/** 실제로 되돌렸는가. needs와 어긋나면 화면이 사용자에게 알린다. */
invalidated_downstream: boolean;
/** 못 만든 구조물 조합 사본 폴더(142-2). */
snapshot_failed?: string[];
}
/** 다른 창이 먼저 저장해 판번호가 어긋났다 — 화면이 최신본을 다시 받아야 한다. */
@@ -231,10 +234,12 @@ export async function saveStructures(
baseRevision: number,
structures: StructureInstance[],
): Promise<StructureSaveResponse> {
return requestJson<StructureSaveResponse>(`/projects/${projectId}/route/structures`, {
method: "PUT",
body: JSON.stringify({ base_revision: baseRevision, structures }),
});
const saved = await requestJson<StructureSaveResponse>(
`/projects/${projectId}/route/structures`,
{ method: "PUT", body: JSON.stringify({ base_revision: baseRevision, structures }) },
);
noteSnapshotFailed(saved.snapshot_failed);
return saved;
}
/** 구 비정규 측점(자유 텍스트)을 구조물 정본으로 이관한다. 멱등 — 배관은 관 지점
B05_Profile/B05_Profile_Structures_Router.py
+8 -4
View File
@@ -14,7 +14,7 @@ from pathlib import Path
from typing import Any
from uuid import UUID
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
@@ -34,6 +34,7 @@ from B05_Profile.B05_Profile_Structures_Schema import (
StructureSaveRequest,
StructureSaveResponse,
)
from common_util.common_util_auth import get_optional_session
from common_util.common_util_drainage_pipes import (
PipePoint,
append_pipe_points_file,
@@ -264,7 +265,9 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
@router.put("/{project_id}/route/structures", response_model=StructureSaveResponse)
async def write_structures(
project_id: UUID, payload: StructureSaveRequest
project_id: UUID,
payload: StructureSaveRequest,
session: dict[str, Any] | None = Depends(get_optional_session), # 가져온이(142-2)
) -> StructureSaveResponse | JSONResponse:
"""구조물 목록을 정본에 덮어쓴다(판번호 불일치 시 409, 타입 오류 시 400)."""
try:
@@ -279,9 +282,9 @@ async def write_structures(
max_chainage_m=await _route_length(project_id),
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
await asyncio.to_thread(safe_sync, root)
failed = await asyncio.to_thread(safe_sync, root, importer(session))
# 설계에 영향을 주는 변경일 때만 B06 이후를 STALE로 돌린다 — 메모만 고쳐도
# 횡단·수량을 다시 돌리게 만들지 않기 위함이다. 응답 플래그는 실제로 STALE
# 전파가 **성공했을 때만** true(실패를 성공처럼 알리지 않는다).
@@ -293,6 +296,7 @@ async def write_structures(
count=len(payload.structures),
needs_downstream_invalidation=needs_invalidation,
invalidated_downstream=invalidated,
snapshot_failed=failed,
)
except LookupError:
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
B05_Profile/B05_Profile_Structures_Schema.py
+2
View File
@@ -279,6 +279,8 @@ class StructureSaveResponse(BaseModel):
# 둘이 어긋나면(필요했는데 못 했다) 화면이 사용자에게 알린다.
needs_downstream_invalidation: bool = False
invalidated_downstream: bool = False
# 못 만든 구조물 조합 복사본 폴더(142-2) — 화면이 [저장] 알림 한 줄
snapshot_failed: list[str] = []
class StructureListResponse(BaseModel):
B08_Quantity/B08_Quantity_Material_Calc.py
+53 -6
View File
@@ -12,6 +12,13 @@
그 밖 ROUNDUP 0 · 131)
구조물 수량 기준 — 일위대가(B09) 재료와 별개 · 두 번 세지 않게 화면 머리에 표시.
단위수량 기본값(143-4) = 연결된 사본 일위대가 자재 · 로직 재료(`summary.snapshot` · 서버가
사본 셈으로 냄 — `B08_Quantity_Router_Material.snapshot_units`) · 자재 열 = 그 사본 자재
(키 = MT 코드 · 로직 재료 = 이름|단위) + 관경 · 2공구 고정 열(`MATERIALS`)은 사용자 값이 있을
때만 · 콘크리트소요 ·
총괄로 끌어옴 = 이름 · 규격 맞춤(`matched`) · 안 맞는 사본 자재 = `unmatched`(화면 알림) ·
사본 없거나 그 자재가 없으면 빈칸 · 관매설 → 파형강관 1 은 그대로 · 손 값이 이김.
사용자 값(`overrides`) = 프로젝트 설정 quantity › material_sheets — 고친 칸만:
{"units": {줄 id: {자재: 값}}, "mix": {배합 줄: {cement|sand|gravel: 값}},
"surcharge": {총괄 줄: %}}
@@ -186,16 +193,38 @@ def diameters(columns: list[dict[str, Any]]) -> list[int]:
return sorted(found)
def material_columns(sizes: list[int]) -> list[tuple[str, str, str, str]]:
def material_columns(
sizes: list[int], shown: set[str] | None = None, extra: list[dict[str, Any]] | None = None
) -> list[tuple[str, str, str, str]]:
"""자재 열 — 고정 열은 `shown`(사용자 값 있음 · None = 전부) · 관경 · 사본 자재(`extra`) 뒤."""
out: list[tuple[str, str, str, str]] = []
for entry in MATERIALS:
if entry[0] == "PIPES":
out += [(f"pipe_{d}", "파형강관", f"Φ{d}", "m") for d in sizes]
else:
elif shown is None or entry[0] in shown:
out.append(entry)
have = {m[0] for m in out}
for m in extra or []:
if m["key"] not in have:
out.append((m["key"], m["name"], m["spec"], m["unit"]))
return out
def _norm(text: str) -> str:
return re.sub(r"\s+", "", str(text or "")).lower()
def matched(name: str, spec: str) -> str | None:
"""사본 자재 → 2공구 고정 열 키(콘크리트소요 · 총괄 끌어옴) — 이름에 고정 이름 · 규격이 다
듦 · 첫 맞춤. ponytail: 글 맞춤 — 품목 이름이 실무 이름과 다르면 「맞춤 없음」(맞춤 표는
필요할 때)."""
text = _norm(f"{name}{spec}")
for key, fixed, fixed_spec, _ in MATERIALS:
if key != "PIPES" and _norm(fixed) in text and _norm(fixed_spec) in text:
return key
return None
def _structure_row(column: dict[str, Any]) -> tuple[str, str, str]:
"""집계표 열 머리 → (구분, 규격, 단위) — 맨 아래 머리가 단위가 아니면(관보호공 모양) 규격에."""
group = [str(g) for g in column.get("group") or []]
@@ -214,8 +243,11 @@ def compute(
columns = summary.get("columns") or []
total = summary.get("total") or {}
units_in = overrides.get("units") or {}
snapshot = summary.get("snapshot") or {}
linked_units = snapshot.get("units") or {}
sizes = diameters(columns)
materials = material_columns(sizes)
shown = {k for cells in units_in.values() for k, v in cells.items() if v is not None}
materials = material_columns(sizes, shown, snapshot.get("materials"))
keys = [m[0] for m in materials]
# 자재수량집계표 — 집계표 열마다 한 줄(커플링은 총괄로 바로)
@@ -228,7 +260,9 @@ def compute(
name, spec, unit = _structure_row(column)
qty = _num(total.get(cid))
length = LENGTH_ID.match(cid)
defaults = {f"pipe_{length.group(1)}": 1.0} if length else {}
defaults = {**(linked_units.get(cid) or {})} # 사본 자재(143-4)
if length:
defaults[f"pipe_{length.group(1)}"] = 1.0
given = units_in.get(cid) or {}
units = {k: _num(_pick(given, k, defaults.get(k))) for k in keys}
amounts: dict[str, float | None] = {}
@@ -248,10 +282,23 @@ def compute(
"amounts": amounts,
}
)
# 끌어옴(143-4) — 고정 · 관경 열은 제 키 · 사본 자재는 이름 · 규격 맞춤 고정 키로
fixed = {m[0] for m in MATERIALS} | {f"pipe_{d}" for d in sizes}
pulled: dict[str, float] = {}
unmatched: list[str] = []
for key, name, spec, _ in materials:
if key not in sums:
continue
target = key if key in fixed else matched(name, spec)
if target is None:
unmatched.append(key)
else:
pulled[target] = round6(pulled.get(target, 0.0) + sums[key])
quantity = {
"materials": [{"key": k, "name": n, "spec": s, "unit": u} for k, n, s, u in materials],
"rows": rows,
"total": sums,
"unmatched": unmatched,
}
# 콘크리트소요자재집계표
@@ -259,7 +306,7 @@ def compute(
concrete_rows = []
concrete_total: dict[str, float] = {}
for key, name, spec, source, mix in CONCRETE_ROWS:
qty = sums.get(source) if source else None
qty = pulled.get(source) if source else None
given = mix_in.get(key) or {}
per = {f: _num(_pick(given, f, mix[f])) for f in MIX_FIELDS}
out = {}
@@ -296,7 +343,7 @@ def compute(
kind, _, name = ref.partition(":")
qty: float | None = None
if kind == "m":
qty = sums.get(name)
qty = pulled.get(name)
elif kind == "c":
qty = concrete_total.get(name)
elif kind == "cp":
B08_Quantity/B08_Quantity_Material_Calc.ts
+57 -8
View File
@@ -11,6 +11,8 @@
* 끝수 = 실무 엑셀 꼴 줄마다(레미콘 ROUND 2 · 철근 · 와이어메쉬 ROUND 3 · 시멘트 ROUND 0 · 그 밖 ROUNDUP 0)
* 사용자 값 = 고친 칸만(키가 있으면 그 값 · null = 빈칸 · 없으면 기본값) — 상수 출처는 파이썬 짝 머리
* 좌측 패널 할증(`panel`) = 총괄 모래 · 자갈 · 시멘트 줄(표 사용자 값보다 앞섬)
* 단위수량 기본값(143-4) = 서버가 낸 사본 자재(`summary.snapshot`) · 자재 열 = 사본 자재 + 관경 ·
* 2공구 고정 열은 사용자 값 있을 때만 · 콘크리트소요 · 총괄 끌어옴 = 이름 · 규격 맞춤(`matched`)
* ========================================================================== */
export type Cells = Record<string, number | null>;
@@ -25,6 +27,8 @@ export interface MaterialOverrides {
export interface SummarySource {
columns: Array<{ id: string; group: string[]; label: string }>;
total: Record<string, number>;
/** 연결된 사본 자재(143-4 · 서버 셈) — 열 · 집계표 열마다 단위수량 기본 */
snapshot?: { materials?: MaterialColumn[]; units?: Record<string, Cells> };
}
export interface MaterialColumn {
@@ -69,7 +73,13 @@ export interface TotalRow {
}
export interface MaterialSheets {
quantity: { materials: MaterialColumn[]; rows: QuantityRow[]; total: Record<string, number> };
quantity: {
materials: MaterialColumn[];
rows: QuantityRow[];
total: Record<string, number>;
/** 2공구 고정 열에 안 맞아 콘크리트소요 · 총괄로 못 끌어온 사본 자재 키(143-4) */
unmatched: string[];
};
concrete: { rows: ConcreteRow[]; total: Record<string, number> };
material: { rows: TotalRow[] };
}
@@ -197,12 +207,35 @@ export function diameters(columns: SummarySource["columns"]): number[] {
return [...found].sort((a, b) => a - b);
}
function materialColumns(sizes: number[]): Entry[] {
return MATERIALS.flatMap((entry): Entry[] =>
/** 자재 열 — 고정 열은 `shown`(사용자 값 있음) · 관경 · 사본 자재(`extra`) 뒤 · 파이썬 짝 */
function materialColumns(
sizes: number[],
shown: Set<string>,
extra: MaterialColumn[] = [],
): Entry[] {
const out = MATERIALS.flatMap((entry): Entry[] =>
entry[0] === "PIPES"
? sizes.map((d): Entry => [`pipe_${d}`, "파형강관", `Φ${d}`, "m"])
: [entry],
: shown.has(entry[0])
? [entry]
: [],
);
const have = new Set(out.map((m) => m[0]));
for (const m of extra) if (!have.has(m.key)) out.push([m.key, m.name, m.spec, m.unit]);
return out;
}
const norm = (text: string): string =>
String(text ?? "")
.replace(/\s+/g, "")
.toLowerCase();
/** 사본 자재 → 2공구 고정 열 키 — 이름에 고정 이름 · 규격이 다 듦 · 첫 맞춤(파이썬 짝 `matched`) */
export function matched(name: string, spec: string): string | null {
const text = norm(`${name}${spec}`);
for (const [key, fixed, fixedSpec] of MATERIALS)
if (key !== "PIPES" && text.includes(norm(fixed)) && text.includes(norm(fixedSpec))) return key;
return null;
}
/** 집계표 열 머리 → [구분, 규격, 단위] — 맨 아래 머리가 단위가 아니면(관보호공 모양) 규격에 붙임 */
@@ -225,8 +258,12 @@ export function computeMaterialSheets(
const columns = summary.columns ?? [];
const total = summary.total ?? {};
const unitsIn = overrides.units ?? {};
const linkedUnits = summary.snapshot?.units ?? {};
const sizes = diameters(columns);
const materials = materialColumns(sizes);
const shown = new Set<string>();
for (const cells of Object.values(unitsIn))
for (const [k, v] of Object.entries(cells)) if (v !== null) shown.add(k);
const materials = materialColumns(sizes, shown, summary.snapshot?.materials);
const keys = materials.map((m) => m[0]);
const rows: QuantityRow[] = [];
@@ -237,7 +274,8 @@ export function computeMaterialSheets(
const [name, spec, unit] = structureRow(column);
const qty = num(total[cid]);
const length = LENGTH_ID.exec(cid);
const defaults: Cells = length ? { [`pipe_${length[1]}`]: 1 } : {};
const defaults: Cells = { ...(linkedUnits[cid] ?? {}) }; // 사본 자재(143-4)
if (length) defaults[`pipe_${length[1]}`] = 1;
const given = unitsIn[cid] ?? {};
const units: Cells = {};
const amounts: Cells = {};
@@ -249,17 +287,28 @@ export function computeMaterialSheets(
}
rows.push({ id: cid, name, spec, unit, qty, units, amounts });
}
// 끌어옴(143-4) — 고정 · 관경 열은 제 키 · 사본 자재는 이름 · 규격 맞춤 고정 키로
const fixed = new Set([...MATERIALS.map((m) => m[0]), ...sizes.map((d) => `pipe_${d}`)]);
const pulled: Record<string, number> = {};
const unmatched: string[] = [];
for (const [key, name, spec] of materials) {
if (!(key in sums)) continue;
const target = fixed.has(key) ? key : matched(name, spec);
if (target === null) unmatched.push(key);
else add(pulled, target, sums[key]);
}
const quantity = {
materials: materials.map(([key, name, spec, unit]) => ({ key, name, spec, unit })),
rows,
total: sums,
unmatched,
};
const mixIn = overrides.mix ?? {};
const concreteRows: ConcreteRow[] = [];
const concreteTotal: Record<string, number> = {};
for (const [key, name, spec, source, mix] of CONCRETE_ROWS) {
const qty = source ? (sums[source] ?? null) : null;
const qty = source ? (pulled[source] ?? null) : null;
const given = mixIn[key] ?? {};
const per: Cells = {};
const amounts: Cells = {};
@@ -306,7 +355,7 @@ export function computeMaterialSheets(
const [kind, name = ""] = ref.split(":");
let qty =
kind === "m"
? (sums[name] ?? null)
? (pulled[name] ?? null)
: kind === "c"
? (concreteTotal[name] ?? null)
: kind === "cp"
B08_Quantity/B08_Quantity_Router_Material.py
+96 -4
View File
@@ -8,10 +8,13 @@
계산 = `B08_Quantity_Material_Calc.py`(화면 짝 `.ts` 와 같은 식) · 총괄 모래 · 자갈 · 시멘트 할증 =
좌측 패널 값(`surcharge_pct` · 저장은 옵션 길 PUT …/quantity/options)
기본값은 파일에 안 씀(읽을 때 채움).
단위수량 기본(143-4) = 연결된 사본 셈(`M02_Item_Copy.calc` = 층 셈 한 벌)의 일위대가 자재 ·
로직 재료(`snapshot_units` · 서버 단독 — 브라우저는 받은 값으로 짝 계산).
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from typing import Any
from uuid import UUID
@@ -20,30 +23,119 @@ from fastapi import APIRouter, Body
from B08_Quantity.B08_Quantity_Material_Calc import (
PANEL_KEY,
_num,
clean_overrides,
compute,
panel_surcharge,
round6,
)
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B08_Quantity.B08_Quantity_Router_Summary import structure_summary
from B08_Quantity.B08_Quantity_Router_Summary import _combo, structure_summary
from common_util.common_util_project_settings import quantity_settings, save_section
from common_util.common_util_summary_item import GROUP_JOIN
router = APIRouter(prefix="/api/projects", tags=["B08 Material"])
SETTINGS_KEY = "material_sheets"
def _folder_units(root: str | Path, folder: str, meta: dict[str, dict]) -> dict[str, float]:
"""사본 하나 = 집계 항목 1 단위의 자재 `{키: 수량}` — 일위대가 자재(키 = MT 코드) + 로직 재료
(키 = 이름|단위 · 재료 수량 × 줄 수량 ÷ 배) · 줄 수량 빈칸 · 0 · 파형강관은 안 담음 · 셈 못 하면
빈 것. ponytail: 참조 줄(승격 ·
사전조합) 안 재료는 안 셈 — 펼친 재료가 셈 답에 오면 그때."""
from M02_MasterTemplete import M02_Item_Copy as copies
from M02_MasterTemplete.M02_Item_Source import PRICE_FILE, read_json
try:
got = copies.calc(root, folder)
bundle = read_json(copies.folder_of(root, folder) / PRICE_FILE) or {}
except (LookupError, ValueError):
return {}
rows = {
row.get("키"): row
for data in bundle.values()
if isinstance(data, dict)
for row in data.get("줄") or data.get("표") or []
if isinstance(row, dict)
}
out: dict[str, float] = {}
def put(key: str, name: str, spec: str, unit: str, qty: float) -> None:
if not qty or name.startswith("파형강관"): # 0 · 관 몫(관매설 → 파형강관 1 · 관경 열)
return
meta.setdefault(key, {"key": key, "name": name, "spec": spec, "unit": unit})
out[key] = out.get(key, 0.0) + qty
for line in got.get("줄") or []:
qty, times = _num(line.get("수량")), _num(line.get("배")) or 1.0
for mat in line.get("재료") or [] if qty is not None else []:
if (per := _num(mat.get("수량"))) is not None:
name, unit = str(mat.get("이름") or ""), str(mat.get("단위") or "")
put(f"{name}|{unit}", name, "", unit, per * qty / times)
for mat in got.get("자재") or []:
code, qty = str(mat.get("코드") or ""), _num(mat.get("수량"))
if code and qty is not None:
row = rows.get(code) or {}
put(
code,
str(row.get("이름") or code),
str(row.get("규격") or ""),
str(row.get("단위") or ""),
qty,
)
return out
def snapshot_units(root: str | Path, summary: dict[str, Any]) -> dict[str, Any]:
"""연결된 사본 자재 → 집계표 열마다 단위수량 기본(143-4) — 칸마다(측점 · 시설) 그 칸을 낸
조합의 사본 자재 × 칸 값을 모아 열 계로 나눔(시설마다 사본이 달라도 계가 맞음) · 사본 없으면
안 담음."""
folders: dict[tuple[str, str | None], list[str]] = {}
for owner_id, entry in (summary.get("linked") or {}).items():
owner = owner_id.partition(GROUP_JOIN)[0]
folders.setdefault((owner, _combo(entry.get("항목"))), []).append(entry["폴더"])
meta: dict[str, dict] = {}
per_folder: dict[str, dict[str, float]] = {}
sums: dict[str, dict[str, float]] = {}
for cid, value, owner, combos in summary.get("sources") or []:
for folder in (f for combo in combos for f in folders.get((owner, combo), [])):
if folder not in per_folder:
per_folder[folder] = _folder_units(root, folder, meta)
cell = sums.setdefault(cid, {})
for key, qty in per_folder[folder].items():
cell[key] = cell.get(key, 0.0) + value * qty
total = summary.get("total") or {}
units = {
cid: {key: round6(qty / total[cid]) for key, qty in cells.items()}
for cid, cells in sums.items()
if cells and total.get(cid)
}
used = {key for cells in units.values() for key in cells}
return {"materials": [m for key, m in meta.items() if key in used], "units": units}
async def _sheets(project_id: UUID, root: str | Path, overrides: dict[str, Any]) -> dict[str, Any]:
summary = await structure_summary(project_id)
source = {"columns": summary["columns"], "total": summary["total"]}
source = {
"columns": summary["columns"],
"total": summary["total"],
"snapshot": await asyncio.to_thread(snapshot_units, root, summary),
}
panel = panel_surcharge(quantity_settings(root).get(PANEL_KEY))
sheets = compute(source, overrides, panel)
notes = list(summary.get("notes") or [])
if lost := sheets["quantity"]["unmatched"]:
notes.append(
f"사본 자재 {len(lost)}종 — 2공구 열 맞춤 없음(콘크리트소요 · 총괄로 안 끌어옴)"
)
return {
"source": source,
"overrides": overrides,
"surcharge_pct": panel,
# 구조물 집계표 알림(「집수정 모양 기록 없음 n곳」 등) — 총괄 · 콘크리트 위 빈 까닭(131)
"notes": list(summary.get("notes") or []),
"sheets": compute(source, overrides, panel),
"notes": notes,
"sheets": sheets,
}
B08_Quantity/B08_Quantity_Router_Summary.py
+1 -67
View File
@@ -47,6 +47,7 @@ from B06_Section.B06_Section_Repository import (
get_workflow_route_context,
)
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B08_Quantity.B08_Quantity_Summary_Sheet import _linked, _Sheet
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_summary_item import GROUP_JOIN
from config.config_db import get_db_pool
@@ -114,21 +115,6 @@ def _combo(item: Any) -> str | None:
return f"{item.get('layer')}|{item.get('owner') or ''}|{item['key']}"
class _Copy(str):
"""잇기 조합 id + 그 사본 주인 id(`linked_snapshots` 키 · `{구조물 id}__{그룹}`) — 145-3."""
copy: str
def _linked(combo: str | None, copy: str) -> str | None:
"""잇기 조합에 사본 주인 id 를 붙임 — 본체 조합(그냥 글)은 시설 id 가 주인."""
if combo is None:
return None
out = _Copy(combo)
out.copy = copy
return out
def one_pipe_length(project_root: str | Path) -> float | None:
"""1본 길이(m) — 프로젝트 층 구조물집계표 변수만(시스템 층으로 안 채움 · 143-2) · 빈칸 None."""
found = layers.read_template(layers.project_dir(project_root), "table", TABLE_NAME)
@@ -187,58 +173,6 @@ def _revet_length(options: dict[str, Any], head: str, unit: str) -> float | None
return (before or 0.0) + (after or 0.0)
class _Sheet:
"""열 · 측점 줄 · 계를 모음 — 같은 측점(1 cm)의 같은 열은 더함."""
def __init__(self) -> None:
self.columns: dict[str, dict[str, Any]] = {}
self.rows: dict[int, dict[str, Any]] = {}
self.owner = "" # 지금 셈하는 시설 — 사본 없어 안 센 칸의 시설을 모음(143-1)
self.unlinked: set[str] = set()
def add(
self,
chainage: float,
key: str,
sort: tuple[Any, ...],
group: list[str],
label: str,
value: float | None,
combos: list[str | None],
) -> None:
if not any(combos): # 그 칸을 낸 조합 사본이 없음 — 안 셈(143-1)
self.unlinked.add(self.owner)
return
cid = f"pipe:{key}"
column = self.columns.setdefault(
cid,
{"id": cid, "group": group, "label": label, "combos": [], "owners": {}, "_sort": sort},
)
for combo in combos:
if combo and combo not in column["combos"]:
column["combos"].append(str(combo))
row = self.rows.setdefault(round(chainage * 100), {"chainage_m": chainage, "values": {}})
if value is not None: # 열은 세움 — 놓였는데 값이 빈칸인 것이 보여야 함
row["values"][cid] = round((row["values"].get(cid) or 0.0) + value, 3)
# 사본 주인별 값(145-3 · 설계내역서가 대가 호표에 이음) — 잇기 사본이 먼저 · 없으면 본체
owner = next((c.copy for c in combos if isinstance(c, _Copy)), self.owner)
owners = column["owners"]
owners[owner] = round(owners.get(owner, 0.0) + value, 3)
def result(self) -> dict[str, Any]:
columns = sorted(self.columns.values(), key=lambda column: column["_sort"])
rows = [self.rows[key] for key in sorted(self.rows)]
total: dict[str, float] = {}
for row in rows:
for cid, value in row["values"].items():
total[cid] = round(total.get(cid, 0.0) + value, 3)
return {
"columns": [{k: v for k, v in c.items() if k != "_sort"} for c in columns],
"rows": rows,
"total": total,
}
def _order(values: tuple[str, ...], value: str) -> int:
return values.index(value) if value in values else len(values)
B08_Quantity/B08_Quantity_Summary_Sheet.py
+78
View File
@@ -0,0 +1,78 @@
"""B08 구조물 집계표 모음 상자 — 열 · 측점 줄 · 계 · 사본 주인별 값 · 칸 출처.
`B08_Quantity_Router_Summary.py` 가 700줄을 넘어 뗌(10-9 통합) — 셈 규칙은 그쪽 그대로.
"""
from __future__ import annotations
from typing import Any
class _Copy(str):
"""잇기 조합 id + 그 사본 주인 id(`linked_snapshots` 키 · `{구조물 id}__{그룹}`) — 145-3."""
copy: str
def _linked(combo: str | None, copy: str) -> str | None:
"""잇기 조합에 사본 주인 id 를 붙임 — 본체 조합(그냥 글)은 시설 id 가 주인."""
if combo is None:
return None
out = _Copy(combo)
out.copy = copy
return out
class _Sheet:
"""열 · 측점 줄 · 계를 모음 — 같은 측점(1 cm)의 같은 열은 더함."""
def __init__(self) -> None:
self.columns: dict[str, dict[str, Any]] = {}
self.rows: dict[int, dict[str, Any]] = {}
self.owner = "" # 지금 셈하는 시설 — 사본 없어 안 센 칸의 시설을 모음(143-1)
self.unlinked: set[str] = set()
self.sources: list[list[Any]] = [] # 칸마다 [열, 값, 시설, 조합] — 자재 재연결(143-4)
def add(
self,
chainage: float,
key: str,
sort: tuple[Any, ...],
group: list[str],
label: str,
value: float | None,
combos: list[str | None],
) -> None:
if not any(combos): # 그 칸을 낸 조합 사본이 없음 — 안 셈(143-1)
self.unlinked.add(self.owner)
return
cid = f"pipe:{key}"
column = self.columns.setdefault(
cid,
{"id": cid, "group": group, "label": label, "combos": [], "owners": {}, "_sort": sort},
)
for combo in combos:
if combo and combo not in column["combos"]:
column["combos"].append(str(combo))
row = self.rows.setdefault(round(chainage * 100), {"chainage_m": chainage, "values": {}})
if value is not None: # 열은 세움 — 놓였는데 값이 빈칸인 것이 보여야 함
row["values"][cid] = round((row["values"].get(cid) or 0.0) + value, 3)
# 사본 주인별 값(145-3 · 설계내역서가 대가 호표에 이음) — 잇기 사본이 먼저 · 없으면 본체
owner = next((c.copy for c in combos if isinstance(c, _Copy)), self.owner)
owners = column["owners"]
owners[owner] = round(owners.get(owner, 0.0) + value, 3)
self.sources.append([cid, value, self.owner, [c for c in combos if c]])
def result(self) -> dict[str, Any]:
columns = sorted(self.columns.values(), key=lambda column: column["_sort"])
rows = [self.rows[key] for key in sorted(self.rows)]
total: dict[str, float] = {}
for row in rows:
for cid, value in row["values"].items():
total[cid] = round(total.get(cid, 0.0) + value, 3)
return {
"columns": [{k: v for k, v in c.items() if k != "_sort"} for c in columns],
"rows": rows,
"total": total,
"sources": self.sources,
}
M02_MasterTemplete/M02_Item_Copy.py
+11 -2
View File
@@ -46,6 +46,7 @@ from M02_MasterTemplete.M02_Item_Source import (
)
#: 길의 종류 이름 → 복사본 파일
PARTS = ("도면", "면적", "일위대가") # 목록 `부분` — 구조물 조합 사본이 가진 것
KINDS = {
"도면": "도면.json",
"면적": "면적.json",
@@ -115,6 +116,9 @@ def list_items(project_root: str | Path) -> list[dict[str, Any]]:
"구조물": manifest.get("구조물") or [],
"고침": bool(manifest.get("고침")),
"값있음": bool(manifest.get("값있음")),
"가져온이": manifest.get("가져온이"),
# 부분(142-2) — 복사본에 있는 파일만(원본 대조 없이)
"부분": [name for name in PARTS if (child / KINDS[name]).is_file()],
}
)
return out
@@ -227,8 +231,11 @@ def _kept_picks(picks: Any, labor: dict[str, Any]) -> dict[str, Any]:
return out
def reapply(project_root: str | Path, folder: str) -> dict[str, Any]:
"""재적용 — 원본(층 · 마스터)을 다시 가져옴 · 입력값 · 고른자재는 같은 키면 살림.
def reapply(
project_root: str | Path, folder: str, who: dict[str, Any] | None = None
) -> dict[str, Any]:
"""재적용 — 원본(층 · 마스터)을 다시 가져옴 · 입력값 · 고른자재는 같은 키면 살림 ·
`who` = 가져온이 새로(142-2).
원본과 달라 덮인 파일(도면 · 면적)은 `덮임` 에 알림."""
target = folder_of(project_root, folder)
@@ -258,6 +265,8 @@ def reapply(project_root: str | Path, folder: str) -> dict[str, Any]:
# 원본을 다시 가져왔으니 시트 고침은 없음 — 살린 입력값 · 고른 자재는 `값있음` 으로 따로 알림
kept = bool(kept_values or files[PICKS_FILE])
fresh = {**manifest, "판": source, "원본판": source, "가져온날": stamp, "고침": False}
if who:
fresh["가져온이"] = who
fresh["값있음"] = kept
fresh.pop("고친날", None)
atomic_write_json(target / MANIFEST_NAME, fresh)
M02_MasterTemplete/M02_Item_Personal.py
+8 -2
View File
@@ -1,7 +1,7 @@
"""프로젝트에서 개인 층으로 — [개인에 저장](save-as) · 프로젝트에서 새 항목(new) (PLAN 26-2b).
개인 층 새 항목 = 구조물집계표 새 열 하나 + 빈 도면 · 면적산출(번호 · SS 키는 저장소가
줌). [개인에 저장]은 구조물 복사본의 열 정의 · 도면 · 면적을 담음 —
줌). [개인에 저장]은 구조물 복사본의 열 정의 · 도면 · 면적 · 일위대가 문서를 담음 —
단가 묶음 · 입력값 · 고른자재는 프로젝트 몫이라 안 가져감. 쓰기는 지금 층 저장소 함수(`store.at` ·
`store.write`) — 나무 고침(`M02_MasterTemplete_Tree`)과
같은 집계표 저장 한 줄기(경로 · 머리 줄 수 맞춤 · 번호 · SS 장부)를 탐(PLAN 26-2a).
@@ -19,7 +19,7 @@ from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_MasterTemplete_Tree as tree
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Source import read_json
from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, read_json
#: 복사본 파일 → 층 양식 종류
PART_KINDS = (("도면.json", "structure"), ("면적.json", "area"))
@@ -89,6 +89,12 @@ def create_item(
body = store._server_values(body)
with store._LOCK:
atomic_write_json(store._path(kind, new_id), body)
labor = (read_json(copy / LABOR_FILE) or {}).get("줄") if copy else None
# 일위대가 문서도(142-2) — 옛 UA 조합 줄은 문서가 아니라 안 담음 · 번호 · 열은 새 열 것
# ponytail: 참조(단가승격 · 사전조합)는 키 그대로 — 개인 층이 못 읽는 참조는 계산 때
# 까닭으로 멈춤(참조사본까지 옮기려면 `Labor_Ref.carry` 길)
if isinstance(labor, dict) and labor.get("종류") == "일위대가":
store.copy_from("labor", new_id, new_id, body=labor)
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
return {"key": ss["키"], "열": new_id, "이름": tree.name_of(heads)}
M02_MasterTemplete/M02_Item_Snapshot.py
+113 -7
View File
@@ -12,6 +12,7 @@
([재적용]은 사용자가)
- 지정을 다른 항목으로 바꾸면 새로 가져옴 · 구조물을 지우거나 지정을 풀면 복사본을 지움 —
프로젝트에서 고친 흔적(`manifest.고침`)이 있으면 지우지 않고 `_지난/` 에 옮겨 둠
- 관 지점을 옮겨도 같은 항목이면 폴더 이름만 바꿈(142-2) · 옵션 그룹 이름 · 가져온이도 굳힘
- 옛 항목 폴더(`templates/items/{층}_{주인}_{키}/` · 25-2b)는 그 항목을 쓰는 구조물마다 복사해
옮겨 받음(고친 값 유지)
"""
@@ -54,6 +55,8 @@ LOCAL_FILES = (VALUES_FILE, PICKS_FILE)
FOLDER_PATTERN = re.compile(
rf"^(?:pipe@\d+(?:\.\d+)?(?:__g[0-9a-z]{{1,16}})?|{STRUCTURE_ID_RULE})$"
)
#: 관 지점 복사본 — (측점, 그룹 꼬리)
PIPE_FOLDER = re.compile(r"^pipe@(\d+(?:\.\d+)?)((?:__g[0-9a-z]{1,16})?)$")
def structures_dir(project_root: str | Path) -> Path:
@@ -221,16 +224,100 @@ def migrate_items(project_root: str | Path) -> list[str]:
return moved
def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
"""구조물 복사본을 지정에 맞춤 — `{만듦, 지움, 못만듦, 옮겨받음}` 폴더 이름."""
def rename_moved(base: Path, wanted: dict[str, tuple[str, dict[str, Any]]]) -> list[str]:
"""관 지점을 옮긴 뒤 [저장] — 새 자리(`pipe@새 측점`)에 복사본이 없고 안 쓰게 된 옛 자리
복사본이 같은 항목 · 같은 그룹 꼬리면 폴더 이름만 바꿈(가까운 측점 먼저 · 142-2) — 가져온날 ·
고침 · 입력값 그대로. 바꾼 것 `옛 이름→새 이름`."""
if not base.is_dir():
return []
def spot(name: str) -> tuple[float, str] | None:
hit = PIPE_FOLDER.match(name)
return (float(hit[1]), hit[2]) if hit else None
new = [
(name, at, item)
for name, (_, item) in wanted.items()
if (at := spot(name)) and not (base / name / MANIFEST_NAME).is_file()
]
old = [
(child.name, at, manifest)
for child in base.iterdir()
if child.is_dir()
and child.name not in wanted
and (at := spot(child.name))
and isinstance(manifest := read_json(child / MANIFEST_NAME), dict)
]
# ponytail: 같은 항목 관 둘을 함께 옮기면 가까운 측점끼리 짝 — 고침 · 입력값이 다를 때만 엇갈림
pairs = sorted(
(abs(a[0] - b[0]), new_name, old_name)
for new_name, a, item in new
for old_name, b, manifest in old
if a[1] == b[1] and same_item(manifest, item)
)
used: set[str] = set()
out: list[str] = []
for _, new_name, old_name in pairs:
target = safe_folder(base, new_name)
if new_name in used or old_name in used or target is None:
continue
if target.is_dir(): # 끊긴 쓰기(manifest 없음) 자리
retire(target)
(base / old_name).rename(target)
manifest = read_json(target / MANIFEST_NAME) or {}
atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": [wanted[new_name][0]]})
used.update((new_name, old_name))
out.append(f"{old_name}→{new_name}")
return out
def group_names() -> dict[str, str]:
"""옵션 그룹 id → 이름(지금 정의) — 복사본 manifest `그룹` 으로 굳힘(142-2)."""
from M02_MasterTemplete.StructOptions.M02_SO_Groups import groups_of
return {g["id"]: g["name"] for rows in groups_of().values() for g in rows}
def copied_groups(
project_root: str | Path | None,
owner: str,
links: dict[str, Any],
defined: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""그 구조물의 옵션 그룹 — 지금 정의 + 복사본에 굳힌 정의(이름 · 사라진 그룹도 · 142-2 ·
층이 바뀌어도 프로젝트 그대로). 굳힌 것 없으면 지금 정의 그대로."""
if project_root is None:
return defined
base = structures_dir(project_root)
ids = {g["id"] for g in defined}
out = []
for group in [*defined, *({"id": gid} for gid in links if gid not in ids)]:
manifest = read_json(
base / folder_name(f"{owner}{GROUP_JOIN}{group['id']}") / MANIFEST_NAME
)
kept = manifest.get("그룹") if isinstance(manifest, dict) else None
if isinstance(kept, dict) and kept.get("name"):
out.append({**group, "name": kept["name"]})
elif group["id"] in ids:
out.append(group)
return out
def sync_snapshots(
project_root: str | Path, who: dict[str, Any] | None = None
) -> dict[str, list[str]]:
"""구조물 복사본을 지정에 맞춤 — `{만듦, 지움, 못만듦, 옮겨받음, 이름바꿈}` 폴더 이름 ·
`who` = 가져온이(manifest · 142-2)."""
moved = migrate_items(project_root)
base = structures_dir(project_root)
wanted = {
folder_name(owner): (owner, item) for owner, item in assignments(project_root).items()
}
renamed = rename_moved(base, wanted)
made: list[str] = []
failed: list[str] = []
removed: list[str] = []
names: dict[str, str] | None = None
for folder, (owner, item) in sorted(wanted.items()):
target = safe_folder(base, folder)
if target is None:
@@ -239,12 +326,20 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
continue
manifest = read_json(target / MANIFEST_NAME)
if isinstance(manifest, dict) and same_item(manifest, item):
if who and not manifest.get("가져온이"): # 미리 가져온 것 — 저장한 이가 가져온이
atomic_write_json(target / MANIFEST_NAME, {**manifest, "가져온이": who})
continue
if target.is_dir(): # 다른 항목으로 바뀜 — 옛 것은 지우거나(고침이면) `_지난` 에
retire(target)
removed.append(folder)
extra: dict[str, Any] = {"가져온이": who} if who else {}
gid = owner.partition(GROUP_JOIN)[2]
if gid:
names = group_names() if names is None else names
if gid in names:
extra["그룹"] = {"id": gid, "name": names[gid]}
try:
write_snapshot(target, item, owner, {})
write_snapshot(target, item, owner, extra)
except (LookupError, ValueError) as error:
logger.warning("집계 항목 복사본을 못 만듦 (%s): %s", folder, error)
failed.append(folder)
@@ -255,17 +350,28 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
if child.is_dir() and not child.name.startswith("_") and child.name not in wanted:
retire(child)
removed.append(child.name)
return {"만듦": made, "지움": removed, "못만듦": failed, "옮겨받음": moved}
out = {"만듦": made, "지움": removed, "못만듦": failed, "옮겨받음": moved}
return {**out, "이름바꿈": renamed} if renamed else out
def safe_sync(project_root: str | Path) -> None:
"""저장 뒤 부름 — 실패해도 저장은 그대로(복사본은 다음 저장이 다시 맞춤)."""
def importer(session: dict[str, Any] | None) -> dict[str, Any] | None:
"""세션 → manifest 가져온이 `{id, 이름}`(142-2)."""
if not isinstance(session, dict) or session.get("user_id") is None:
return None
return {"id": int(session["user_id"]), "이름": session.get("name")}
def safe_sync(project_root: str | Path, who: dict[str, Any] | None = None) -> list[str]:
"""저장 뒤 부름 — 실패해도 저장은 그대로(복사본은 다음 저장이 다시 맞춤) · 못 만든 복사본
폴더 이름을 돌려줌(저장 응답 → [저장] 알림 · 142-2 · 통째 실패 = `["*"]`)."""
try:
result = sync_snapshots(project_root)
result = sync_snapshots(project_root, who)
if any(result.values()):
logger.info("집계 항목 복사본: %s", result)
return result["못만듦"]
except Exception:
logger.exception("집계 항목 복사본 맞춤 실패 (저장은 유지): %s", project_root)
return ["*"]
def complete_old(folder: Path) -> bool:
M02_MasterTemplete/M02_Labor_Ref.py
+15 -2
View File
@@ -422,9 +422,11 @@ class Expand:
from M02_MasterTemplete import M02_Labor_Doc as labor
lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)]
kept = doc.get("알약") # 프로젝트 참조사본 = 가져온 때 굳힌 알약(142-2) · 층 = 지금 M01
pills = kept if isinstance(kept, list) else None
by_logic = {
code: pill
for pill in combo_store.combo_pills({"담은로직": lines})["물량"]
for pill in pills or combo_store.combo_pills({"담은로직": lines})["물량"]
for code in pill["로직"]
}
inner, rows, why = [], [], None
@@ -478,13 +480,24 @@ def frozen(doc: dict[str, Any]) -> dict[str, dict[str, Any]]:
continue
got = read_ref(ref)
if got is not None:
out[ref_id(ref)] = got
out[ref_id(ref)] = with_pills(got)
walk(got.get("담은로직"), depth + 1)
walk(doc.get("담은로직"), 1)
return out
def with_pills(doc: dict[str, Any]) -> dict[str, Any]:
"""사전조합 굳힘 — 안 로직 줄 알약(배 · 단위)도 그때 M01 값으로(142-2 · 사본 셈이 지금 M01 을
안 읽게)."""
if doc.get("종류") != TITLE["precombo"]:
return doc
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)]
return {**doc, "알약": combo_store.combo_pills({"담은로직": lines})["물량"]}
def frozen_logics(copies: dict[str, Any]) -> list[str]:
"""굳힌 단가승격 · 사전조합이 담은 로직 키 — 프로젝트 단가 묶음에 넣음."""
return [
M02_MasterTemplete/M02_Project_Items_Router.py
+3 -1
View File
@@ -29,6 +29,7 @@ from common_util.common_util_summary_item import parse_summary_item
from M02_MasterTemplete import M02_Item_Copy as copies
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layer_routes
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete.M02_Item_Snapshot import importer
from M02_MasterTemplete.M02_Item_Source import item_name
router = APIRouter(prefix="/api/projects", tags=["M02 Project Items"])
@@ -162,7 +163,8 @@ async def write_item_file(
async def reapply_item(
project_id: str, folder: str, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
return await _run(copies.reapply, await _root(session, project_id), folder)
root = await _root(session, project_id)
return await _run(copies.reapply, root, folder, importer(session)) # 가져온이(142-2)
@router.post("/{project_id}/items/{folder}/save-as")
M02_MasterTemplete/M02_Table_Fill.py
+7 -5
View File
@@ -140,14 +140,15 @@ def collect_items(
"group_items": point.group_items,
}
)
return _with_group_items(items)
return _with_group_items(items, project_root)
def _with_group_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
def _with_group_items(items: list[dict[str, Any]], project_root=None) -> list[dict[str, Any]]:
"""옵션 그룹(M02 PLAN 46)마다 한 건 더 — 같은 구조물 · 옵션 · 자리에 그 그룹이 고른 항목 ·
복사본 `{구조물 id}__{그룹 id}`. 정의된 그룹만(미지정 그룹 = 「미지정」 알림에 셈) · 정의가 사라진
그룹 연결은 기록에만 남고 표에 안 들어감."""
복사본 `{구조물 id}__{그룹 id}`. 그룹 = 지금 정의 + 복사본에 굳힌 정의(142-2 · 이름 · 사라진
그룹도) · 미지정 그룹 = 「미지정」 알림에 셈."""
from common_util.common_util_summary_item import GROUP_JOIN
from M02_MasterTemplete.M02_Item_Snapshot import copied_groups
from M02_MasterTemplete.M02_Table_Fill_Spec import container_rows
from M02_MasterTemplete.StructOptions.M02_SO_Groups import groups_of
@@ -156,7 +157,8 @@ def _with_group_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
for item in items:
links = item.pop("group_items", None) or {}
out.append(item)
for group in defined.get(item["type_id"], []):
mine = defined.get(item["type_id"], [])
for group in copied_groups(project_root, item["복사본"], links, mine):
out.append(
{
**item,
M02_MasterTemplete/StructOptions/M02_SO_Style.css
+11
View File
@@ -357,3 +357,14 @@
.m02-sod__specpreview {
max-width: 28rem;
}
/* 미리보기 상자 이름표 [삭제] — B05 상자 안 작은 단추 틀과 같은 꼴(136-3 · M02 는 B05 CSS 를 안 읽음) */
.m02-sod__block .b05-structure__group > legend > button {
box-sizing: border-box;
height: 26px;
min-height: 0;
padding-block: 0;
padding-inline: var(--spacing-8);
font-size: var(--text-caption);
line-height: 1;
}
resources/tester/test_121_b08_material_sheets.py
+3 -2
View File
@@ -104,7 +104,8 @@ def test_자재수량_계가_엑셀_66행과_같다():
assert "pipe:cp|800" not in rows
revet = rows["pipe:revet|돌쌓기(찰)|2.0"]
assert revet["qty"] == 20 and revet["amounts"]["stone_buy_45"] == 38.4
assert revet["amounts"]["sod"] is None # 빈 단위수량 = 빈칸(0 아님)
assert revet["amounts"]["rubble"] == 15 and revet["amounts"]["stone_pick_45"] is None # 빈칸
assert "sod" not in revet["amounts"] # 2공구 고정 열 = 사용자 값 · 사본 있을 때만(143-4)
basin = rows["pipe:basin|800|U"]
assert (basin["name"], basin["spec"], basin["unit"]) == (
"관 공",
@@ -112,7 +113,7 @@ def test_자재수량_계가_엑셀_66행과_같다():
"개소",
)
keys = [m["key"] for m in sheets["quantity"]["materials"]]
assert keys.index("rebar_d25") + 1 == keys.index("pipe_800")
assert keys.index("concrete_machine") + 1 == keys.index("pipe_800")
assert keys.index("pipe_1200") + 1 == keys.index("mortar_1_3")
resources/tester/test_136_3_remove_button.py
+7
View File
@@ -42,3 +42,10 @@ def test_이름표_단추는_작은_단추_틀_흐림은_글에만():
assert "height: var(--b05-control-h)" in small and "padding-block: 0" in small
legend = _rule(".b05-structure__group > legend {")
assert "opacity" not in legend and "currentColor 80%" in legend
def test_M02_미리보기_이름표_단추도_작은_단추_틀():
css = (ROOT / "M02_MasterTemplete" / "StructOptions" / "M02_SO_Style.css").read_text("utf-8")
at = css.index(".m02-sod__block .b05-structure__group > legend > button")
rule = css[at : css.index("}", at)]
assert "height: 26px" in rule and "padding-block: 0" in rule
resources/tester/test_143_4_b08_snapshot_materials.py
+117
View File
@@ -0,0 +1,117 @@
"""143-4 사본 일위대가 자재 → B08 자재수량집계표 단위수량 재연결.
- 서버 `snapshot_units` — 칸(측점 · 시설)마다 그 칸을 낸 조합의 연결된 사본 셈(자재 · 로직 재료) ×
칸 값 ÷ 열 계 · 사본 없거나 줄 수량 빈칸이면 빈칸(안 담음)
- 짝 계산 `compute` — 열 = 사본 자재 + 관경 · 2공구 고정 열은 사용자 값 있을 때만 · 손 값이 이김 ·
관매설 → 파형강관 1 그대로 · 콘크리트소요 · 총괄 끌어옴 = 이름 · 규격 맞춤 · 안 맞으면 `unmatched`
- 거울 = TS 짝(`B08_Quantity_Material_Calc.ts`)이 같은 답
"""
from __future__ import annotations
import json
from B08_Quantity import B08_Quantity_Router_Material as router
from B08_Quantity.B08_Quantity_Material_Calc import clean_overrides, compute
from M02_MasterTemplete import M02_Item_Copy as copies
from M02_MasterTemplete import M02_Item_Snapshot as snap
from resources.tester.test_121_b08_material_sheets import _run_ts
from resources.tester.test_item_copy_complete import ( # noqa: F401 — 고정물
FOLDER,
PIPE,
_pipe,
_save,
master_copy,
world,
)
COMBO = "system||SS000011"
LEN = "pipe:len|1000"
SUMMARY = {
"columns": [
{"id": LEN, "group": ["관 공", "Φ1000m/m", "관매설"], "label": "m"},
{"id": "pipe:revet|돌쌓기(찰)|2.0", "group": ["흙막이", "찰쌓기", "H=2.0"], "label": "m"},
{"id": "pipe:chute|메붙임", "group": ["돌붙임", "메붙임"], "label": "㎡"},
],
"total": {LEN: 20, "pipe:revet|돌쌓기(찰)|2.0": 10, "pipe:chute|메붙임": 5},
"snapshot": {
"materials": [
{"key": "MT000001", "name": "레미콘", "spec": "25-18-80", "unit": "㎥"},
{"key": "고임돌|㎥", "name": "고임돌", "spec": "채집", "unit": "㎥"},
{"key": "야면석|㎥", "name": "야면석", "spec": "", "unit": "㎥"},
],
"units": {
"pipe:revet|돌쌓기(찰)|2.0": {"MT000001": 0.31, "고임돌|㎥": 0.2, "야면석|㎥": 1.5},
"pipe:chute|메붙임": {"고임돌|㎥": 0.12},
},
},
}
def test_사본_자재가_열과_단위수량_기본():
over = clean_overrides({"units": {"pipe:chute|메붙임": {"고임돌|㎥": 0.5, "sod": 1}}})
sheets = compute(SUMMARY, over)
keys = [m["key"] for m in sheets["quantity"]["materials"]]
assert keys == ["pipe_1000", "sod", "MT000001", "고임돌|㎥", "야면석|㎥"] # 고정 열 = 손 값만
rows = {row["id"]: row for row in sheets["quantity"]["rows"]}
revet = rows["pipe:revet|돌쌓기(찰)|2.0"]
assert revet["units"]["MT000001"] == 0.31 and revet["amounts"]["MT000001"] == 3.1
assert revet["units"]["pipe_1000"] is None # 사본에 없는 자재 = 빈칸
assert rows["pipe:chute|메붙임"]["units"]["고임돌|㎥"] == 0.5 # 손 값이 이김
assert rows[LEN]["units"]["pipe_1000"] == 1 # 관매설 → 파형강관 1 그대로
# 끌어옴 — 레미콘 25-18-80 · 고임돌(채집) = 이름 · 규격 맞춤 · 야면석은 맞춤 없음
assert sheets["quantity"]["unmatched"] == ["야면석|㎥"]
total = {row["key"]: row for row in sheets["material"]["rows"]}
assert total["remicon_25_18_80"]["qty"] == 3.1
assert total["pipe_1000"]["qty"] == 20
concrete = {row["key"]: row for row in sheets["concrete"]["rows"]}
assert concrete["c40"]["qty"] is None # 콘크리트(기계) 사본 없음 = 빈칸
def test_거울_TS와_파이썬이_같다():
over = clean_overrides({"units": {"pipe:chute|메붙임": {"고임돌|㎥": 0.5, "sod": 1}}})
assert _run_ts(SUMMARY, over) == compute(SUMMARY, over)
bare = {k: v for k, v in SUMMARY.items() if k != "snapshot"}
assert _run_ts(bare, clean_overrides({})) == compute(bare, clean_overrides({}))
def test_서버_사본_셈_자재_x_칸값_나누기_열계(world): # noqa: F811
"""일위대가 자재(MT 코드 · 손 수량 2) = 단위수량 2 · 사본 없는 시설 칸은 안 셈."""
root = world["root"]
_save(root, [_pipe(), {**_pipe(None), "chainage_m": 80.0}])
folder = root / "templates/structures" / FOLDER
labor = json.loads((folder / "일위대가.json").read_text("utf-8"))
labor["줄"] = {
"종류": "일위대가",
"열": "x",
"단위": "m",
"담은로직": [{"로직": "GF000176"}], # 줄 수량 빈칸 — 로직 재료 안 셈
"자재": [{"코드": "MT003810", "수량": {"손": 2}}],
}
(folder / "일위대가.json").write_text(json.dumps(labor, ensure_ascii=False), "utf-8")
summary = {
"linked": snap.linked_snapshots(root),
"sources": [[LEN, 12.0, FOLDER, [COMBO]], [LEN, 8.0, "pipe@80.0", [COMBO]]],
"total": {LEN: 20.0},
}
got = router.snapshot_units(root, summary)
assert got["units"] == {LEN: {"MT003810": 1.2}} # 12 × 2 ÷ 20 · 80 m 관은 사본 없음
assert [m["key"] for m in got["materials"]] == ["MT003810"]
assert got["materials"][0]["name"] != "MT003810" # 이름 = 사본 단가 묶음 품목 이름
def test_로직_재료는_재료_x_줄수량_나누기_배(monkeypatch):
fake = {
"줄": [
{"수량": 4, "배": 10, "재료": [{"이름": "돌", "단위": "㎥", "수량": 5}]},
{"수량": None, "배": 1, "재료": [{"이름": "돌", "단위": "㎥", "수량": 9}]}, # 빈칸
{"수량": 1, "배": 1, "재료": [{"이름": "파형강관", "단위": "m", "수량": 1}]}, # 관 몫
],
"자재": [],
}
monkeypatch.setattr(copies, "calc", lambda root, folder: fake)
monkeypatch.setattr(copies, "folder_of", lambda root, folder: copies.Path("없음"))
meta: dict = {}
assert router._folder_units("r", "f", meta) == {"돌|㎥": 2.0}
assert meta == {"돌|㎥": {"key": "돌|㎥", "name": "돌", "spec": "", "unit": "㎥"}}
resources/tester/test_b05_structures_router.py
+11
View File
@@ -83,6 +83,17 @@ def test_save_then_read_roundtrip(client):
assert read.json()["structures"][0]["structure_id"]
def test_save_response_carries_snapshot_failed(client, monkeypatch):
"""142-2 — 못 만든 구조물 조합 사본을 저장 응답에 실음(화면 [저장] 알림) · 로그인 세션 없으면
가져온이 없음."""
from M02_MasterTemplete import M02_Item_Snapshot as snap
seen = []
monkeypatch.setattr(snap, "safe_sync", lambda root, who: seen.append(who) or ["st9"])
saved = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
assert saved.json()["snapshot_failed"] == ["st9"] and seen == [None]
def test_stale_revision_returns_409(client):
client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
conflict = client.put(
resources/tester/test_item_copy_complete.py
+18
View File
@@ -135,6 +135,8 @@ def test_지정하면_새_파일까지_복사(world):
"구조물": ["pipe@40.0"],
"고침": False,
"값있음": False,
"가져온이": None,
"부분": ["도면", "면적", "일위대가"], # 142-2 — 사본이 가진 부분
}
]
@@ -428,6 +430,9 @@ def test_재적용은_시트_고침을_덮고_알림_입력값은_살림(world):
assert "메모" not in copies.read_file(root, FOLDER, "면적")["내용"]
assert copies.read_file(root, FOLDER, "입력값")["내용"] == values
assert copies.reapply(root, FOLDER)["덮임"] == []
who = {"id": 42, "이름": "재적용"} # 142-2 — [재적용]이 가져온이를 새로 적음
copies.reapply(root, FOLDER, who)
assert copies.list_items(root)[0]["가져온이"] == who
def test_개인에_저장하면_개인_층에_항목(world):
@@ -453,6 +458,19 @@ def test_개인에_저장하면_개인_층에_항목(world):
assert _save_all(root, [_wall("b1", made["summary_item"])], [])["만듦"] == ["b1"]
def test_개인에_저장은_일위대가_문서도_가져감_142_2(world):
root = world["root"]
_save(root, [_pipe()])
copy = root / "templates/structures" / FOLDER
doc = {"종류": "일위대가", "열": "old", "단위": "m", "담은로직": [{"로직": "GF000176"}]}
labor = {**_read(copy / "일위대가.json"), "줄": doc}
(copy / "일위대가.json").write_text(json.dumps(labor, ensure_ascii=False), "utf-8")
made = copies.save_as(root, FOLDER, 3, 12, ["관공"], "문서 담음")
got = _read(layers.personal_dir(3, 12) / "labor" / f"{made['열']}.json")
assert got["열"] == made["열"] and got["일위대가번"] # 번호 · 열은 새 열 것
assert [line["로직"] for line in got["담은로직"]] == ["GF000176"]
def test_프로젝트에서_새_항목(world):
root = world["root"]
_save_all(root, [_wall("c1")], [])
resources/tester/test_m02_labor_ref.py
+25
View File
@@ -496,3 +496,28 @@ def test_서고_구분_상세구분_사전조합은_고른_값_옛_문서는_읽
rows = [{"종류": "precombo", "이름": sc}]
refs.summaries(world["system"], rows)
assert (rows[0]["구분"], rows[0]["상세구분"], rows[0]["연결"]) == ("가", "나", None)
def test_사전조합_굳힘은_안_로직_알약도_사본_값_142_2(world, monkeypatch):
"""프로젝트 참조사본 사전조합 = 가져온 때 알약(배 10) — 그 뒤 M01 이 바뀌어도 사본 셈 그대로 ·
층 문서(알약 없음)는 지금 M01 을 읽음."""
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
from M02_MasterTemplete import M02_Item_Prices as prices
from M02_MasterTemplete import M02_Item_Source as source
sc = _combo(world["system"], [{"로직": TEN}], "열배")
ref = {**SYS, "키": sc}
with store.at(world["system"]):
doc = labor.clean({"열": "rv", "담은로직": [{"참조": ref, "수량": {"손": 1}}]})
rows = source.labor_doc_rows(doc)
kept = rows["참조사본"][refs.ref_id(ref)]["알약"]
assert [(p["로직"], p.get("배")) for p in kept] == [([TEN], 10)]
assert "알약" not in _read(world["system"], "precombo", sc) # 층 문서는 그대로
bundle = prices.to_json(prices.collect([TEN]))
before = prices.calc(rows, bundle, {}, {}, None)["줄"][0]["기준"]["계"]
changed = {"물량": [{"키": "x|㎡", "이름": "x", "단위": "㎡", "로직": [TEN]}]} # 배 1 로 바뀜
monkeypatch.setattr(combo_store, "combo_pills", lambda row: changed)
assert prices.calc(rows, bundle, {}, {}, None)["줄"][0]["기준"]["계"] == pytest.approx(before)
with store.at(world["system"]):
live = labor.calc(doc, {})["줄"][0]["기준"]["계"]
assert live == pytest.approx(before * 10) # 층 셈은 지금 M01(배 1)
resources/tester/test_structure_group_items.py
+27
View File
@@ -95,3 +95,30 @@ def test_채움은_정의된_그룹마다_한_건_더(world, monkeypatch):
items = fill.collect_items(root)
assert [i["복사본"] for i in items] == ["st0", "st0__g1", "st1", "st1__g1"]
assert items[1]["그룹"] == "기슭막이" and items[1]["summary_item"] == RV
def test_그룹_정의도_사본_값_정의가_바뀌거나_사라져도_그대로(world, monkeypatch):
"""142-2 — 그룹 연결 사본 manifest `그룹` = 가져온 때 정의(이름) · 이름을 바꾸거나 그룹을
지워도 프로젝트 집계 줄은 사본 이름 그대로 · 연결 없는 구조물은 지금 정의만."""
root, system = world["root"], world["system"]
_define(system, monkeypatch, {"masonry_dry": [{"id": "g1", "이름": "기슭막이", "옵션": []}]})
linked = {**_dry(MS), "group_items": {"g1": RV}}
bare = {**_dry(None), "chainage_m": 125, "start_m": 120, "end_m": 130}
_save(root, [linked, bare], [])
manifest = json.loads((_items(root) / "st0__g1/manifest.json").read_text("utf-8"))
assert manifest["그룹"] == {"id": "g1", "name": "기슭막이"}
_define(system, monkeypatch, {"masonry_dry": [{"id": "g1", "이름": "바뀐 이름", "옵션": []}]})
items = fill.collect_items(root)
assert [(i["복사본"], i.get("그룹")) for i in items] == [
("st0", None),
("st0__g1", "기슭막이"),
("st1", None),
("st1__g1", "바뀐 이름"),
]
_define(system, monkeypatch, {})
items = fill.collect_items(root)
assert [(i["복사본"], i.get("그룹")) for i in items] == [
("st0", None),
("st0__g1", "기슭막이"),
("st1", None),
]
resources/tester/test_structures_summary_snapshot.py
+49
View File
@@ -218,6 +218,55 @@ def test_원본에_없는_항목은_못만듦으로_남김(world):
assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["st0"]
def _manifest(root: Path, folder: str) -> dict[str, Any]:
return json.loads((_items(root) / folder / "manifest.json").read_text("utf-8"))
def test_관_옮겨_저장해도_같은_사본_이름만_바꿈(world):
"""142-2 — 같은 항목 관을 옮기면 폴더 이름만(가져온날 · 고침 · 입력값 그대로 · 원본을 다시
안 읽음) · 같은 항목 관 둘 = 가까운 측점끼리 · 다른 항목 자리는 새로."""
root, system = world["root"], world["system"]
_save(root, [], [_pipe(PIPE), {**_pipe(PIPE), "chainage_m": 100.0}])
for folder in ("pipe@40.0", "pipe@100.0"):
manifest = {**_manifest(root, folder), "고침": True, "가져온날": f"옛날-{folder}"}
(_items(root) / folder / "manifest.json").write_text(json.dumps(manifest), "utf-8")
(_items(root) / "pipe@40.0/입력값.json").write_text('{"L": {"x": {"값": 7}}}', "utf-8")
(system / layers.COMBO_FILE).unlink() # 원본이 사라져도 이름만 바꾸므로 못만듦 없음
moved = [{**_pipe(PIPE), "chainage_m": 98.0}, {**_pipe(PIPE), "chainage_m": 41.5}]
result = _save(root, [], moved)
assert sorted(result["이름바꿈"]) == ["pipe@100.0→pipe@98.0", "pipe@40.0→pipe@41.5"]
assert result["만듦"] == result["지움"] == result["못만듦"] == []
assert sorted(p.name for p in _items(root).iterdir()) == ["pipe@41.5", "pipe@98.0"]
near = _manifest(root, "pipe@41.5")
assert near["가져온날"] == "옛날-pipe@40.0" and near["고침"] is True
assert near["구조물"] == ["pipe@41.5"]
assert json.loads((_items(root) / "pipe@41.5/입력값.json").read_text("utf-8"))["L"]["x"]
assert _manifest(root, "pipe@98.0")["가져온날"] == "옛날-pipe@100.0"
def test_관_옮김_다른_항목이면_이름_안_바꿈(world):
root = world["root"]
_save(root, [], [_pipe(PIPE)])
result = _save(root, [], [{**_pipe(MS), "chainage_m": 41.5}])
assert "이름바꿈" not in result
assert result["만듦"] == ["pipe@41.5"] and result["지움"] == ["pipe@40.0"]
assert _manifest(root, "pipe@41.5")["원본"]["key"] == MS["key"]
def test_못만듦은_저장_응답으로_가져온이는_manifest_에(world):
"""142-2 — `safe_sync` 가 못 만든 폴더를 돌려줌(저장 응답 → [저장] 알림) · 가져온이 =
저장한 세션 · 미리 가져온 사본(가져온이 없음)에도 [저장] 때 채움."""
root = world["root"]
_save(root, [_dry({**MS, "key": "SS999999"}), _dry(MS)], [])
who = snap.importer({"user_id": 42, "name": "검증"})
assert who == {"id": 42, "이름": "검증"} and snap.importer(None) is None
assert snap.safe_sync(root, who) == ["st0"]
assert _manifest(root, "st1").get("가져온이") == who # 앞 저장(가져온이 없음) 사본도 채움
shutil.rmtree(_items(root) / "st1")
snap.safe_sync(root, who)
assert _manifest(root, "st1")["가져온이"] == who
# ── 채움 ─────────────────────────────────────────────