perf(B09 [저장] · 708-1): 같은 입력 되풀이 셈 묶음 — 화면 [저장] 3.7 s → 1.7 s · 결과 바이트 같음
- M01 캐시 `steady()` 범위 — 요청 한 번 안 파일 자국 한 번(77 번 × 19 ms → 1 번) · stale · forget 뒤 다시 · 범위 밖 그대로 - 두 번 셈(183-1) 2차 산근 = 1차가 본 단가(있음 · 값)가 같을 때만 1차 풀이 사본 · 다르면 다시 셈 - [저장] 단가 수요 한 번(셈 답에 넘김) · 단계 표시 = 셈과 함께 - B08 `once()` 범위 — B09 원천 읽기 때 토공 채집석 공제가 자재 집계표 답을 다시 씀(두 번 셈 → 한 번) - 시험: test_b09_save_speed 새로 · test_183 바꿔 끼운 셈 함수 = 새 memo 인자 받음 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -11,11 +11,14 @@
|
||||
단위수량 기본(143-4) = 연결된 사본 셈(`M02_Item_Copy.calc` = 층 셈 한 벌)의 일위대가 자재 ·
|
||||
로직 재료(`snapshot_units` · 서버 단독 — 브라우저는 받은 값으로 짝 계산) · 참조 줄(승격 · 사전조합)
|
||||
안 재료도 셈 · 기계 주연료 · 할증(ℓ)은 뺌(원가 몫 · 160-3).
|
||||
`once()` 범위 = 저장본 세 표 한 번(B09 원천 읽기 — 토공집계표 채집석 공제가 같은 답을 씀).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import contextvars
|
||||
import re
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
@@ -231,10 +234,27 @@ async def _sheets(project_id: UUID, root: str | Path, overrides: dict[str, Any])
|
||||
}
|
||||
|
||||
|
||||
#: `once()` 범위 — {프로젝트 id: 저장본 세 표}(`material_sheets` 가 남김 · `picked_deduction` 이 씀)
|
||||
_SAVED: contextvars.ContextVar[dict | None] = contextvars.ContextVar("b08_saved", default=None)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def once():
|
||||
"""저장본이 안 바뀌는 범위 — 저장본 세 표를 한 번만 셈."""
|
||||
token = _SAVED.set({})
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_SAVED.reset(token)
|
||||
|
||||
|
||||
async def picked_deduction(project_id: UUID, root: str | Path) -> float | None:
|
||||
"""채집석 공제 ㎥(160-5) — 저장본 기준 총괄 「채집석 공제」 줄 값 · 토공 잔토처리에서 뺌."""
|
||||
overrides = clean_overrides(quantity_settings(root).get(SETTINGS_KEY))
|
||||
rows = (await _sheets(project_id, root, overrides))["sheets"]["material"]["rows"]
|
||||
got = (_SAVED.get() or {}).get(project_id)
|
||||
if got is None:
|
||||
overrides = clean_overrides(quantity_settings(root).get(SETTINGS_KEY))
|
||||
got = await _sheets(project_id, root, overrides)
|
||||
rows = got["sheets"]["material"]["rows"]
|
||||
return next((row["amount"] for row in rows if row["key"] == DEDUCT_KEY), None)
|
||||
|
||||
|
||||
@@ -242,7 +262,10 @@ async def picked_deduction(project_id: UUID, root: str | Path) -> float | None:
|
||||
async def material_sheets(project_id: UUID) -> dict[str, Any]:
|
||||
root = await _project_root(project_id)
|
||||
overrides = clean_overrides(quantity_settings(root).get(SETTINGS_KEY))
|
||||
return await _sheets(project_id, root, overrides)
|
||||
got = await _sheets(project_id, root, overrides)
|
||||
if (saved := _SAVED.get()) is not None:
|
||||
saved[project_id] = got
|
||||
return got
|
||||
|
||||
|
||||
@router.put("/{project_id}/quantity/materials")
|
||||
|
||||
@@ -69,6 +69,7 @@ from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone
|
||||
from B08_Quantity.B08_Quantity_Router_Earthwork import earthwork_table
|
||||
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
|
||||
from B08_Quantity.B08_Quantity_Router_Material import material_sheets
|
||||
from B08_Quantity.B08_Quantity_Router_Material import once as material_once
|
||||
from B08_Quantity.B08_Quantity_Router_Milestone import read_data as read_milestone
|
||||
from B08_Quantity.B08_Quantity_Router_Milestone import site_info
|
||||
from B09_Estimation import B09_Estimation_Bill_Library as library
|
||||
@@ -95,6 +96,7 @@ from common_util.common_util_auth import verify_session
|
||||
from common_util.common_util_project_settings import reset_section
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from common_util.common_util_workflow_state import mark_stage
|
||||
from M01_MasterData import M01_MasterData_Store_Cache as m01_cache
|
||||
from M02_MasterTemplete import M02_Layer_Access as access
|
||||
from M02_MasterTemplete.M02_Item_Snapshot import importer
|
||||
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import _call, _root
|
||||
@@ -128,17 +130,19 @@ async def _b08_for(project_id: UUID, notes: list[str], reuse: bool) -> dict[str,
|
||||
|
||||
|
||||
async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
|
||||
"""B08 수량 원천 — 못 읽으면 빈칸 + 알림(B09 는 B08 를 고치지 않음)."""
|
||||
"""B08 수량 원천 — 못 읽으면 빈칸 + 알림(B09 는 B08 를 고치지 않음) ·
|
||||
자재 집계표 한 번(토공 채집석 공제가 같은 답 · `material_once`)."""
|
||||
out: dict[str, Any] = {}
|
||||
for key, name, read in (
|
||||
("materials", "자재 집계표", material_sheets),
|
||||
("earthwork", "토공집계표", earthwork_table),
|
||||
):
|
||||
try:
|
||||
out[key] = await read(project_id)
|
||||
except Exception as error: # noqa: BLE001 — 원천 하나가 막혀도 나머지 표는 그림
|
||||
notes.append(f"B08 {name} 못 읽음 — {getattr(error, 'detail', error)}")
|
||||
out[key] = None
|
||||
with material_once():
|
||||
for key, name, read in (
|
||||
("materials", "자재 집계표", material_sheets),
|
||||
("earthwork", "토공집계표", earthwork_table),
|
||||
):
|
||||
try:
|
||||
out[key] = await read(project_id)
|
||||
except Exception as error: # noqa: BLE001 — 원천 하나가 막혀도 나머지 표는 그림
|
||||
notes.append(f"B08 {name} 못 읽음 — {getattr(error, 'detail', error)}")
|
||||
out[key] = None
|
||||
# 구조물 집계표 = 자재 집계표가 이미 부른 값(열 · 계 · 조합) — 두 번 안 부름
|
||||
out["summary"] = (out["materials"] or {}).get("source")
|
||||
# 총괄 · 콘크리트 표 = B08 답의 `sheets` 한 층 아래(재료비목록표 · 운반 수량이 읽음 · 167-2)
|
||||
@@ -146,12 +150,15 @@ async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
|
||||
return out
|
||||
|
||||
|
||||
def compute_sheets(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]:
|
||||
def compute_sheets(
|
||||
root: Path, inputs: dict[str, Any], user: dict[str, Any], memo: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""짝 셈 한 벌 — 바탕(144) → 산근(145-1 · 서버 단독) → 대가(145-2) → 내역(145-3) → 총괄(145-4).
|
||||
|
||||
화면 짝 = `B09_Estimation_UI_Shell.computeSheets`(산근만 서버 답 그대로)."""
|
||||
화면 짝 = `B09_Estimation_UI_Shell.computeSheets`(산근만 서버 답 그대로) ·
|
||||
memo = 두 번 셈의 앞 산근 풀이(`Sangeun.collect`)."""
|
||||
base = compute_base(inputs, user)
|
||||
sg = sangeun.build(root, base, user) # 145-1 — 굳힌 단가로 풀이(+ 166-2 손 줄 로직)
|
||||
sg = sangeun.build(root, base, user, memo) # 145-1 — 굳힌 단가로 풀이(+ 166-2 손 줄 로직)
|
||||
by_key = {row["key"]: row for row in sg["sg_list"]}
|
||||
labor = compute_labor(
|
||||
inputs.get("daega"), {**base, "gaps": []}, by_key, user.get("daega_cover")
|
||||
@@ -171,13 +178,15 @@ def compute_sheets(root: Path, inputs: dict[str, Any], user: dict[str, Any]) ->
|
||||
def compute_twice(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> tuple[dict, dict]:
|
||||
"""183-1 두 번 셈 — 1차 셈으로 쓰는 자재만 남기고(`used_materials`) 재료비목록표 수량 =
|
||||
1차 역전개 총수량(`material_qty` = 재료비수량금액집계표 수량) → 2차 셈 · 집계표.
|
||||
`inputs` 를 고쳐 둠(화면 짝이 거른 입력 · 수량을 그대로 받음)."""
|
||||
first = compute_sheets(root, inputs, user)
|
||||
`inputs` 를 고쳐 둠(화면 짝이 거른 입력 · 수량을 그대로 받음) ·
|
||||
2차 산근 = 1차가 본 단가가 같으면 1차 풀이 그대로(`memo`)."""
|
||||
memo: dict[str, Any] = {}
|
||||
first = compute_sheets(root, inputs, user, memo)
|
||||
summary = compute_summary(first, hopyo_lines(first))
|
||||
keep = used_materials(first, inputs, user)
|
||||
inputs["materials"] = [m for m in inputs.get("materials") or [] if m["key"] in keep]
|
||||
inputs["material_qty"] = {row["key"]: row["qty"] for row in summary["sum_material"]}
|
||||
sheets = compute_sheets(root, inputs, user)
|
||||
sheets = compute_sheets(root, inputs, user, memo)
|
||||
return sheets, compute_summary(sheets, hopyo_lines(sheets))
|
||||
|
||||
|
||||
@@ -218,11 +227,27 @@ async def _demand(root: Path, draft: dict[str, Any] | None = None) -> list[str]:
|
||||
|
||||
|
||||
async def payload(
|
||||
project_id: UUID, root: Path, draft: dict[str, Any] | None = None, reuse_b08: bool = False
|
||||
project_id: UUID,
|
||||
root: Path,
|
||||
draft: dict[str, Any] | None = None,
|
||||
reuse_b08: bool = False,
|
||||
keys: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""셈 답 한 벌 — keys = 방금 센 단가 수요(같은 저장 값 · [저장]) · 마스터 자국은 한 번만."""
|
||||
with m01_cache.steady():
|
||||
return await _payload(project_id, root, draft, reuse_b08, keys)
|
||||
|
||||
|
||||
async def _payload(
|
||||
project_id: UUID,
|
||||
root: Path,
|
||||
draft: dict[str, Any] | None,
|
||||
reuse_b08: bool,
|
||||
keys: list[str] | None,
|
||||
) -> dict[str, Any]:
|
||||
notes: list[str] = []
|
||||
b08 = await _b08_for(project_id, notes, reuse_b08)
|
||||
keys = await _demand(root, draft)
|
||||
keys = await _demand(root, draft) if keys is None else keys
|
||||
if not keys:
|
||||
notes.append(
|
||||
"연결된 구조물 조합 사본 없음 — 노무 · 경비 · 중기 단가 줄은 사본이 연결되면 생김"
|
||||
@@ -268,6 +293,11 @@ async def estimation(project_id: UUID) -> dict[str, Any]:
|
||||
@router.put("/{project_id}/estimation")
|
||||
async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
|
||||
root = await _project_root(project_id)
|
||||
with m01_cache.steady():
|
||||
return await _save(project_id, root, body)
|
||||
|
||||
|
||||
async def _save(project_id: UUID, root: Path, body: dict[str, Any]) -> dict[str, Any]:
|
||||
user = await asyncio.to_thread(store.write_user, root, body.get("user"))
|
||||
calc = await asyncio.to_thread(calc_user, root) # 263-2 운반 값 사본 = 지금 거리
|
||||
if (rows := calc["bill"].get("rows")) != ((user.get("bill") or {}).get("rows") or []):
|
||||
@@ -279,8 +309,11 @@ async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) ->
|
||||
book = await asyncio.to_thread(store.read_book, root)
|
||||
book = await asyncio.to_thread(prices.extend_book, book, keys)
|
||||
await asyncio.to_thread(store.write_book, root, book)
|
||||
await mark_stage(str(project_id), 6, "save") # [저장](B1)
|
||||
return await payload(project_id, root)
|
||||
# [저장](B1) 단계 표시는 셈과 함께 · 수요 = 방금 센 것(저장 값 · 사본 그대로)
|
||||
out, _ = await asyncio.gather(
|
||||
payload(project_id, root, keys=keys), mark_stage(str(project_id), 6, "save")
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
@router.post("/{project_id}/estimation/complete")
|
||||
|
||||
@@ -30,6 +30,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
@@ -148,13 +149,49 @@ def rows_of(folder: Path, whole: Any = None) -> list[dict[str, Any]]:
|
||||
|
||||
|
||||
# ── 풀이 ──────────────────────────────────────────────────────────────
|
||||
class _Seen(dict):
|
||||
"""읽은 키를 적는 단가 표 — 두 번 셈(183-1)에서 첫 풀이가 본 값만 견줌."""
|
||||
|
||||
def __init__(self, *args: Any) -> None:
|
||||
super().__init__(*args)
|
||||
self.seen: set[Any] = set()
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
self.seen.add(key)
|
||||
return super().__contains__(key)
|
||||
|
||||
def __getitem__(self, key: Any) -> Any:
|
||||
self.seen.add(key)
|
||||
return super().__getitem__(key)
|
||||
|
||||
def get(self, key: Any, default: Any = None) -> Any:
|
||||
self.seen.add(key)
|
||||
return super().get(key, default)
|
||||
|
||||
def same(self, other: dict) -> bool:
|
||||
"""본 키마다 있음 · 값이 `other` 와 같음."""
|
||||
return all(
|
||||
dict.__contains__(self, k) == dict.__contains__(other, k)
|
||||
and dict.get(self, k) == dict.get(other, k)
|
||||
for k in self.seen
|
||||
)
|
||||
|
||||
|
||||
class _Prices:
|
||||
"""B09 표 단가 — 노무 · 중기(노 · 재 · 경) · 자재 적용 단가(키 → 값)."""
|
||||
|
||||
def __init__(self, sheets: dict[str, Any]) -> None:
|
||||
self.labor = {r["key"]: num(r.get("price")) for r in sheets.get("labor") or []}
|
||||
self.machine = {r["key"]: r for r in sheets.get("machine") or []}
|
||||
self.material = {r["key"]: num(r.get("applied")) for r in sheets.get("compare") or []}
|
||||
self.labor = _Seen({r["key"]: num(r.get("price")) for r in sheets.get("labor") or []})
|
||||
self.machine = _Seen({r["key"]: r for r in sheets.get("machine") or []})
|
||||
self.material = _Seen(
|
||||
{r["key"]: num(r.get("applied")) for r in sheets.get("compare") or []}
|
||||
)
|
||||
|
||||
def same(self, other: _Prices) -> bool:
|
||||
"""첫 풀이가 본 단가가 `other` 에서도 같음 — 같으면 풀이도 같음."""
|
||||
return all(
|
||||
getattr(self, k).same(getattr(other, k)) for k in ("labor", "machine", "material")
|
||||
)
|
||||
|
||||
|
||||
def _unit(code: str) -> tuple[str, Decimal]:
|
||||
@@ -538,10 +575,14 @@ def collect(
|
||||
entries: list[tuple[str, Path]],
|
||||
sheets: dict[str, Any],
|
||||
hands: list[tuple[str, str, dict[str, Any]]] | None = None,
|
||||
memo: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""`entries` = [(구조물 id, 사본 폴더)] → 블록 · 목록 · 「연결 없음」 ·
|
||||
`hands` = 손 줄 로직(166-2 · 사본 뒤 번호 · 마스터 로직)."""
|
||||
`hands` = 손 줄 로직(166-2 · 사본 뒤 번호 · 마스터 로직) ·
|
||||
memo = 같은 사본 · 손 줄의 앞 풀이(두 번 셈) — 앞 풀이가 본 단가가 같으면 그 풀이 사본."""
|
||||
solver = Solver(sheets)
|
||||
if memo and memo["prices"].same(solver.prices):
|
||||
return copy.deepcopy(memo["sg"])
|
||||
for owner, folder in entries:
|
||||
whole = item_prices.load_prices(read_json(folder / PRICE_FILE) or {})
|
||||
for src in rows_of(folder, whole):
|
||||
@@ -558,20 +599,27 @@ def collect(
|
||||
for b in blocks
|
||||
if b["sum"] is None
|
||||
]
|
||||
return {
|
||||
out = {
|
||||
"sg": blocks,
|
||||
"sg_list": listing,
|
||||
"sg_gaps": gaps,
|
||||
"sg_notes": [NO_SOURCE] + ([] if blocks else [EMPTY]),
|
||||
"sg_hand": hand_keys,
|
||||
}
|
||||
if memo is not None:
|
||||
memo.update(prices=solver.prices, sg=copy.deepcopy(out))
|
||||
return out
|
||||
|
||||
|
||||
def build(
|
||||
project_root: str | Path, sheets: dict[str, Any], user: dict[str, Any] | None = None
|
||||
project_root: str | Path,
|
||||
sheets: dict[str, Any],
|
||||
user: dict[str, Any] | None = None,
|
||||
memo: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""연결된 사본 전부(구조물 id 차례 → 설계내역서 서고 사본 · 702-3)의 산근 +
|
||||
설계내역서 손 줄 로직(166-2 · 뒤 번호) — 차례 = `Labor_Source.daega_inputs`."""
|
||||
설계내역서 손 줄 로직(166-2 · 뒤 번호) — 차례 = `Labor_Source.daega_inputs` ·
|
||||
memo = 같은 root · user 두 번 셈의 앞 풀이(`collect`)."""
|
||||
from B09_Estimation.B09_Estimation_Bill_Library import copies
|
||||
|
||||
base = structures_dir(project_root)
|
||||
@@ -583,4 +631,4 @@ def build(
|
||||
entries += [
|
||||
(f"서고 {(m.get('원본') or {}).get('name') or f.name}", f) for f, m in copies(project_root)
|
||||
]
|
||||
return collect(entries, sheets, hand_sources(user))
|
||||
return collect(entries, sheets, hand_sources(user), memo)
|
||||
|
||||
@@ -10,6 +10,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import contextvars
|
||||
import json
|
||||
import sys
|
||||
import threading
|
||||
@@ -28,6 +30,28 @@ _INDEX: dict = {} # 폴더 · 자국 · files · whole
|
||||
_BLOCK: dict[str, tuple[tuple, dict]] = {} # 로직 파일 → (자국, {키: `_check` 가 센 것})
|
||||
_BASE: dict[str, tuple] = {} # 막힘을 셀 때 바탕이던, 로직 아닌 파일의 자국
|
||||
_MARKED: dict[str, set] = {} # 우리가 고친 요소·표 파일 → 흔들린 자리(`narrow` 가 고른 것)
|
||||
#: `steady()` 범위 — 폴더 → 자국(범위 안에선 자국 한 번 · `to_thread` 도 같은 묶음)
|
||||
_STEADY: contextvars.ContextVar[dict | None] = contextvars.ContextVar("m01_steady", default=None)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def steady():
|
||||
"""요청 한 번처럼 마스터가 안 바뀌는 범위 — `loaded` 가 파일 자국을 한 번만 셈."""
|
||||
token = _STEADY.set({}) if _STEADY.get() is None else None
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if token is not None:
|
||||
_STEADY.reset(token)
|
||||
|
||||
|
||||
def _now(folder: Path) -> dict[str, tuple]:
|
||||
held = _STEADY.get()
|
||||
if held is None:
|
||||
return _stamps(folder)
|
||||
if folder not in held:
|
||||
held[folder] = _stamps(folder)
|
||||
return held[folder]
|
||||
|
||||
|
||||
def _stamps(folder: Path) -> dict[str, tuple]:
|
||||
@@ -84,6 +108,8 @@ def forget() -> None:
|
||||
_BLOCK.clear()
|
||||
_BASE.clear()
|
||||
_MARKED.clear()
|
||||
if (held := _STEADY.get()) is not None:
|
||||
held.clear()
|
||||
|
||||
|
||||
def stale(names, marks=()) -> None:
|
||||
@@ -97,6 +123,8 @@ def stale(names, marks=()) -> None:
|
||||
if was.get("그룹") != "로직":
|
||||
_MARKED.setdefault(name, set()).update(marks)
|
||||
_INDEX.pop("자국", None)
|
||||
if (held := _STEADY.get()) is not None:
|
||||
held.clear()
|
||||
|
||||
|
||||
def loaded(folder: Path) -> tuple[dict[str, dict], object]:
|
||||
@@ -105,7 +133,7 @@ def loaded(folder: Path) -> tuple[dict[str, dict], object]:
|
||||
if _INDEX.get("폴더") != folder: # 시험은 사본 폴더로 바꿈
|
||||
forget()
|
||||
_INDEX["폴더"] = folder
|
||||
now = _stamps(folder)
|
||||
now = _now(folder)
|
||||
for gone in [name for name in _READ if name not in now]:
|
||||
del _READ[gone]
|
||||
for name, stamp in now.items():
|
||||
|
||||
@@ -59,7 +59,7 @@ def test_수량_원천_material_qty_거울():
|
||||
def test_두_번_셈_후보_줄_빠지고_수량_역전개(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def fake(root, inputs, user):
|
||||
def fake(root, inputs, user, memo=None):
|
||||
calls.append([m["key"] for m in inputs["materials"]])
|
||||
base = compute_base(inputs, user)
|
||||
bill = {"rows": [{"kind": "item", "qty": 3, "ref": {"tab": "sg", "key": "SG1"}}]}
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""B09 [저장] 속도 — 같은 결과 범위 안 다시 셈 줄임 지킴.
|
||||
|
||||
- 산근 두 번 셈(183-1) = 1차가 본 단가(있음 · 값)가 같을 때만 1차 풀이 그대로(`_Prices.same`)
|
||||
- M01 `steady()` 범위 = 파일 자국 한 번 · `stale` 뒤 다시 셈 · 범위 밖은 늘 셈
|
||||
- B08 `once()` 범위 = 채집석 공제가 자재 집계표 답을 다시 씀(셈 한 번)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from B08_Quantity import B08_Quantity_Router_Material as material # noqa: E402
|
||||
from B09_Estimation.B09_Estimation_Sangeun import _Prices # noqa: E402
|
||||
from M01_MasterData import M01_MasterData_Store_Cache as cache # noqa: E402
|
||||
|
||||
SHEETS = {
|
||||
"labor": [{"key": "L1", "price": 100}, {"key": "L2", "price": 5}],
|
||||
"compare": [{"key": "M1", "applied": 3}],
|
||||
"machine": [{"key": "E1", "labor": 1, "material": 2, "expense": 3}],
|
||||
}
|
||||
|
||||
|
||||
def _seen() -> _Prices:
|
||||
got = _Prices(SHEETS)
|
||||
got.labor.get("L1")
|
||||
assert "M9" not in got.material
|
||||
got.machine.get("E1")
|
||||
return got
|
||||
|
||||
|
||||
def test_산근_같은_단가면_다시_씀() -> None:
|
||||
other = {**SHEETS, "labor": [{"key": "L1", "price": 100}, {"key": "L2", "price": 999}]}
|
||||
assert _seen().same(_Prices({**other, "compare": []})) # 안 본 L2 · M1 은 안 셈
|
||||
|
||||
|
||||
def test_산근_본_단가_바뀌면_다시_셈() -> None:
|
||||
assert not _seen().same(_Prices({**SHEETS, "labor": [{"key": "L1", "price": 101}]}))
|
||||
added = {**SHEETS, "compare": [{"key": "M9", "applied": 1}]}
|
||||
assert not _seen().same(_Prices(added)) # 없던 자재가 생김
|
||||
machine = {**SHEETS, "machine": [{"key": "E1", "labor": 1, "material": 2, "expense": 4}]}
|
||||
assert not _seen().same(_Prices(machine))
|
||||
|
||||
|
||||
def test_M01_범위_안_자국_한_번(monkeypatch) -> None:
|
||||
calls = []
|
||||
monkeypatch.setattr(cache, "_stamps", lambda folder: calls.append(folder) or {})
|
||||
folder = Path("x")
|
||||
cache._now(folder)
|
||||
cache._now(folder)
|
||||
assert len(calls) == 2 # 범위 밖 = 늘 셈
|
||||
with cache.steady():
|
||||
cache._now(folder)
|
||||
cache._now(folder)
|
||||
assert len(calls) == 3
|
||||
cache.stale([]) # 스스로 고친 자리 = 다시 셈
|
||||
cache._now(folder)
|
||||
assert len(calls) == 4
|
||||
cache._now(folder)
|
||||
assert len(calls) == 5
|
||||
|
||||
|
||||
def test_B08_범위_안_자재_집계표_한_번(monkeypatch, tmp_path) -> None:
|
||||
calls = []
|
||||
rows = [{"key": material.DEDUCT_KEY, "amount": 12.5}]
|
||||
|
||||
async def sheets(_pid, _root, _over):
|
||||
calls.append(1)
|
||||
return {"sheets": {"material": {"rows": rows}}}
|
||||
|
||||
async def root(_pid):
|
||||
return tmp_path
|
||||
|
||||
monkeypatch.setattr(material, "_sheets", sheets)
|
||||
monkeypatch.setattr(material, "_project_root", root)
|
||||
monkeypatch.setattr(material, "quantity_settings", lambda _root: {})
|
||||
|
||||
async def run() -> list:
|
||||
out = [await material.picked_deduction("p", tmp_path)]
|
||||
with material.once():
|
||||
await material.material_sheets("p")
|
||||
out.append(await material.picked_deduction("p", tmp_path))
|
||||
out.append(await material.picked_deduction("p", tmp_path))
|
||||
return out
|
||||
|
||||
assert asyncio.run(run()) == [12.5, 12.5, 12.5]
|
||||
assert len(calls) == 3 # 범위 안 공제 = 셈 없음
|
||||
Reference in New Issue
Block a user