Merge remote-tracking branches 'origin/sub_laptop_4' and 'origin/sub_laptop_1' into main_laptop_1

This commit is contained in:
2026-10-15 12:38:16 +09:00
18 changed files with 1236 additions and 51 deletions
@@ -350,6 +350,18 @@ async def run_auto_design_chain(
logger.exception("초기값 스냅샷 실패: project_id=%s", project_id)
mark_design_failed(project_root, f"초기값 스냅샷 저장 실패: {exc}")
# 노선 추천 기본 다섯 알약 미리 셈(PLAN 327-2) — 예약만 · 체인이 [초기화] 요청 안에서도
# 불리므로 기다리지 않는다. 같은 입력이면 저장본이 있어 셈 없이 끝난다.
try:
from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task
from B05_Profile.B05_Profile_Router_RouteInitial import precompute_defaults
_schedule_background_task(
precompute_defaults(project_id), task_name=f"추천 노선 미리 셈 {project_id}"
)
except Exception: # noqa: BLE001 — 미리 셈이 체인을 막지는 않는다
logger.exception("추천 노선 미리 셈 예약 실패: project_id=%s", project_id)
return {
"route_id": route_id,
"length_m": float(solve_result.total_length_m or 0.0),
+20 -6
View File
@@ -3,8 +3,9 @@
예상노선 추종 · 영선 노선 · 토공 최소 · 기울기 순한 · 안전 우선 · 계류 횡단 최소 ·
공사비 어림 최소 · 비중.
계산 자리 = 서버 단독(③). 결과는 화면 캐시용 — 이 모듈은 파일을 쓰지 않는다. 적용은 화면이
받은 노드로 기존 `POST /route/replan`([확인])을 부른다.
계산 자리 = 서버 단독(③). 이 모듈은 파일을 쓰지 않는다 — 결과는 라우터가 입력 지문 키로
`B05_Profile/route_recommend/` 에 남긴다(`..._Store` · PLAN 327). 적용은 화면이 받은 노드로
기존 `POST /route/replan`([확인])을 부른다.
갈래(`mode`):
· follow — 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 규칙 · 같은 선.
@@ -182,6 +183,20 @@ def write_initial_choice(
return choice
def resolve_inputs(
max_grade: float,
stream_offset_m: float | None,
grade_smooth_m: float | None,
target_grade: float | None,
) -> tuple[float, float, float]:
"""(계류 이격, 기울기 잴 지반 폭, 영선 목표 기울기) — 비운 값은 기본값 · 저장 키와 같은 풀이."""
stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m)
grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m)
# 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만)
goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade)
return stream_offset, grade_smooth, goal_grade
def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]:
"""갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값."""
if mode != "weighted":
@@ -240,10 +255,9 @@ def generate_initial_route(
if len(expected) < 2:
raise ValueError("예상노선 정점이 2개 미만입니다.")
used = resolve_weights(mode, weights)
stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m)
grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m)
# 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만)
goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade)
stream_offset, grade_smooth, goal_grade = resolve_inputs(
max_grade, stream_offset_m, grade_smooth_m, target_grade
)
def measure(
outline: Any, *, bound: float | None, follow: bool = False, weighted: bool = False
@@ -1,8 +1,9 @@
"""노선 추천 후보의 균형 종단(PLAN 63-4 · 63-2 안 A).
후보 노선 하나에 종단을 세워 절 · 성토를 **횡단 ㎥** 로 맞춘 계획선과 그 토량을 낸다.
편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 파일 · 캐시 · initial_snapshot · 확정 종단을
바꾸지 않는다. 확정 체인의 종단 규칙(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다.
편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 작업본 파일 · initial_snapshot · 확정 종단을
바꾸지 않는다(추천 결과째 저장본에 남는 것은 라우터 몫 · PLAN 327). 확정 체인의 종단 규칙
(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다.
계산 자리 = 서버 단독(③) — 화면은 받은 측점 · 지반고 · 계획고를 그리기만 한다.
@@ -0,0 +1,128 @@
"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3).
같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면
셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 상한 없음).
층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리
`B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때
다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임).
"""
from __future__ import annotations
import asyncio
import hashlib
import json
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Any, AsyncIterator
from B05_Profile.B05_Profile_Engine_RouteInitial import (
DEFAULT_WEIGHTS,
Terrain,
_pool_key,
resolve_inputs,
resolve_weights,
)
from common_util.common_util_json import atomic_write_json
ALGO_VERSION = 1
STORE_DIR = "B05_Profile/route_recommend"
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
_LOCKS: dict[str, asyncio.Lock] = {}
def store_dir(project_root: Path) -> Path:
return Path(project_root) / STORE_DIR
def request_fingerprint(
expected: list[tuple[float, float]],
terrain: Terrain,
streams: list[list[tuple[float, float]]],
*,
mode: str,
weights: dict[str, float] | None,
max_grade: float,
min_radius_m: float,
stream_offset_m: float | None,
grade_smooth_m: float | None,
target_grade: float | None,
with_profile: bool,
design_speed: int,
) -> tuple[str, str]:
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
pool = _pool_key(
expected,
terrain,
streams,
max_grade,
min_radius_m,
*resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
).hex()
used = resolve_weights(mode, weights)
tail = (mode, sorted(used.items()), bool(with_profile), int(design_speed), ALGO_VERSION)
key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
return key, pool
def is_default(
mode: str, weights: dict[str, float] | None, criteria: dict[str, Any] | None, terrain: Any
) -> bool:
"""기본값 결과 — 빈 제약 · 50/25/25 · 설계 지형. 표시일 뿐 지우는 데 쓰지 않음."""
empty = not any(value is not None for value in (criteria or {}).values())
plain = mode != "weighted" or resolve_weights(mode, weights) == DEFAULT_WEIGHTS
return empty and plain and terrain is None
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
"""저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
try:
data = json.loads((store_dir(project_root) / f"{key}.json").read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
return None
return data
def write_stored(
project_root: Path,
key: str,
pool: str,
request: dict[str, Any],
result: dict[str, Any],
default: bool,
) -> bool:
"""새 키만 적는다 — 이미 있으면 그대로 두고 False(덮어쓰기 없음)."""
path = store_dir(project_root) / f"{key}.json"
if path.exists():
return False
atomic_write_json(
path,
{
"version": ALGO_VERSION,
"key": key,
"pool": pool,
"default": default,
"created_at": datetime.now().isoformat(timespec="seconds"),
"request": request,
"result": result,
},
)
return True
@asynccontextmanager
async def pool_lock(pool: str) -> AsyncIterator[None]:
# ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
async with _LOCKS.setdefault(pool, asyncio.Lock()):
yield
+1 -1
View File
@@ -26,7 +26,7 @@
GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계)
POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산
POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(파일 안 씀)
POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀)
"""
import asyncio
+87 -23
View File
@@ -30,14 +30,35 @@ async def initial_route(
) -> dict[str, Any] | JSONResponse:
"""초기 계획노선 갈래 하나를 만들어 비교값과 함께 돌려준다(PLAN 20장).
**파일을 쓰지 않는다** — 화면 캐시용. 적용은 받은 `nodes` 로 `POST /route/replan`([확인]).
작업본은 쓰지 않는다 — 추천 저장본(`..._Store` · PLAN 327)만. 적용은 받은 `nodes` 로
`POST /route/replan`([확인]).
"""
return await compute_initial(project_id, request)
async def compute_initial(
project_id: UUID, request: replan.RouteInitialRequest
) -> dict[str, Any] | JSONResponse:
"""같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
"""
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
from B05_Profile.B05_Profile_Engine_RouteInitial import (
generate_initial_route,
load_terrain_and_streams,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import project_pipes
from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
cover_shortfall,
project_pipes,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
is_default,
pool_lock,
read_stored,
request_fingerprint,
write_stored,
)
from common_util.common_util_surface_confirmation import (
analysis_terrain_params,
get_surface_confirmation_params,
@@ -81,37 +102,62 @@ async def initial_route(
async with pool.acquire() as connection:
selection = await get_surface_confirmation_params(connection, str(project_id))
selection = analysis_terrain_params(selection, request.terrain)
inputs = {
"mode": request.mode,
"weights": weights,
"max_grade": max_grade,
"min_radius_m": radius_m,
"stream_offset_m": stream_offset_m,
"grade_smooth_m": grade_smooth_m,
"target_grade": None if target_pct is None else target_pct / 100.0,
"design_speed": design_speed or 20,
"with_profile": request.with_profile,
}
started = time.perf_counter()
try:
terrain, streams = await asyncio.to_thread(
load_terrain_and_streams, project_root, selection
)
pipes = (
await asyncio.to_thread(project_pipes, project_root) if request.with_profile else None
)
result = await asyncio.to_thread(
functools.partial(
generate_initial_route,
[(x, y) for x, y in expected],
terrain,
streams,
mode=request.mode,
weights=weights,
max_grade=max_grade,
min_radius_m=radius_m,
stream_offset_m=stream_offset_m,
grade_smooth_m=grade_smooth_m,
target_grade=None if target_pct is None else target_pct / 100.0,
design_speed=design_speed or 20,
pipes=pipes,
with_profile=request.with_profile,
)
route = [(x, y) for x, y in expected]
key, pool_key = await asyncio.to_thread(
functools.partial(request_fingerprint, route, terrain, streams, **inputs)
)
async with pool_lock(pool_key):
stored = await asyncio.to_thread(read_stored, project_root, key)
if stored is not None:
result = stored["result"]
else:
# 관 지점은 키 밖 — 덮개 수 없이 셈해 남기고 아래에서 지금 관으로 센다.
result = await asyncio.to_thread(
functools.partial(
generate_initial_route, route, terrain, streams, pipes=None, **inputs
)
)
await asyncio.to_thread(
write_stored,
project_root,
key,
pool_key,
request.model_dump(),
result,
is_default(
request.mode,
weights,
criteria.model_dump() if criteria else None,
request.terrain,
),
)
profile = result["metrics"].get("profile")
if request.with_profile and profile and "error" not in profile:
pipes = await asyncio.to_thread(project_pipes, project_root)
if pipes:
profile.update(cover_shortfall(profile, result["planned"], pipes))
except (FileNotFoundError, ValueError) as error:
return JSONResponse(status_code=409, content={"status": "error", "message": str(error)})
logger.info(
"초기 계획노선 %s: project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
"초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm",
request.mode,
"저장본" if stored is not None else "셈",
project_id,
time.perf_counter() - started,
result["metrics"]["length_m"],
@@ -122,10 +168,28 @@ async def initial_route(
"status": "success",
"project_id": str(project_id),
**result,
"stored": stored is not None,
"current_initial": current,
}
async def precompute_defaults(project_id: UUID) -> None:
"""업로드 체인 끝 예약(PLAN 327-2) — 기본 다섯 알약을 미리 셈해 남김 · 있으면 꺼내기만."""
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES
started = time.perf_counter()
for mode in DEFAULT_MODES:
# 화면 기본과 같은 몸통 — 빈 제약 · 50/25/25 · 설계 지형 · 추적만 종단 끔(`withProfileOf`)
request = replan.RouteInitialRequest(mode=mode, with_profile=mode != "follow")
outcome = await compute_initial(project_id, request)
if isinstance(outcome, JSONResponse):
logger.warning(
"추천 미리 셈 멈춤(%s %s): project_id=%s", mode, outcome.status_code, project_id
)
return
logger.info("추천 미리 셈 끝: project_id=%s %.1fs", project_id, time.perf_counter() - started)
def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int:
"""종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다."""
import json
@@ -189,6 +189,8 @@ export interface InitialChoice {
similar?: SimilarRoute[] | null;
/** 제약 안에서 갈 수 있는 길의 띠 폭 어림(m) — 서버 `band_width_m`. */
bandWidthM?: number | null;
/** 서버 추천 저장본에서 꺼낸 결과인가(계획서 327) — 서버 `stored`. */
stored?: boolean;
/** 균형 종단(계획서 63-4) — 모달 미리 보기에만. 캐시에는 안 적는다(`writeInitialChoice`). */
balance?: BalanceProfile | null;
}
@@ -336,6 +338,7 @@ export function normalizeRouteInitial(
terrainBoundPct: body.relaxed === true ? num(body.terrain_bound_pct) : null,
similar: parseSimilar(body.similar),
bandWidthM: num(body.band_width_m),
stored: body.stored === true,
balance: parseBalance(m.profile),
criteria: body.criteria
? {
@@ -462,7 +462,9 @@ export function openRecommendModal(params: RecommendParams): void {
preview = choice;
if (choice.balance) seen.set(label, choice.balance);
else seen.delete(label); // 종단 없는 갈래(예상노선 추적)
stateEl.textContent = `미리 보기: ${choiceLabel(logic, sentWeights)} — [적용]을 눌러야 편집 창에 놓입니다.`;
stateEl.textContent = `미리 보기: ${choiceLabel(logic, sentWeights)}${
choice.stored ? " (저장된 결과)" : ""
} — [적용]을 눌러야 편집 창에 놓입니다.`;
} catch (error) {
if (closed || mine !== seq) return;
stateEl.textContent = "셈하지 못했습니다.";
+231
View File
@@ -0,0 +1,231 @@
"""M02 일위대가 실무자료 단가산출 근거(PLAN 330) — 엑셀 「단가산출」 블록 → 출처 · 직종 ·
우리 돈 로직과 견준 근거 후보.
읽기(`read`) = B열(머리 「산출근거」) 머리 줄 「N.이름[단위]」(앞 빈칸 없음)부터
다음 머리 앞까지 한 블록 ·
출처 = 「…품셈」(오타 「품줌」) 뒤 20 글자 안 절 번호 — 앞 글에 산림 · 임도 = 산림품셈 ·
그 밖 = 건설품셈 ·
직종 = 「이름 : 수」 줄의 빈칸 뺀 이름이 우리 인력(LB) 이름 · 옛이름일 때만(지금 이름으로).
견줌(`candidates`) = 같은 책 조항 같음 1.0 · 앞 마디 같음 0.5(조항이 안 맞으면 후보 아님) ·
건설 공통 · 토목 덤 0.1 · 직종 겹침(보통인부 뺌) 0.5 · 단위 0.15 · 직종도 뜻 낱말도 안 겹치면 버림 ·
같은 점수는 글자 두 개 묶음(괄호 글 포함)이 가름. 자동 확정 없음 — 맞춤 차례는 `M02_Hopyo_Match`.
M01 · 원본 로직은 읽기만.
"""
from __future__ import annotations
import re
from functools import lru_cache
from typing import Any
from M01_MasterData import M01_MasterData_Store as m01
from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match # 서로 부름 — 부를 때만 씀
from M02_MasterTemplete import M02_Labor_Inputs as labor_inputs
SHEET = "단가산출"
COMMON = "보통인부"
BONUS_PARTS = ("공통", "토목")
_HEAD = re.compile(r"^(\d+)\.\s*(.*?)\s*\[([^\[\]]*)\]\s*$")
_BOOK = re.compile(r"품[셈줌]")
_TRADE = re.compile(r"^[\s*·\-]*([가-힣][가-힣\s]*?)\s*:\s*[\d,]")
_TYPOS = (("깍기", "깎기"),)
_MONEY = {"합계": "계", "노무비": "노무비", "재료비": "재료비", "경비": "경비"}
def _flat(value: Any) -> str:
return re.sub(r"\s+", "", str(value or ""))
# ── 엑셀 블록 읽기 ─────────────────────────────────────────────────────
@lru_cache(maxsize=2)
def _trade_names(whole_id: int) -> dict[str, str]:
"""{빈칸 뺀 이름 · 옛이름: 지금 이름} — 우리 인력(LB)."""
out: dict[str, str] = {}
for row in m01.loaded()[1].numbers.get("LB", {}).values():
name = str(row.get("이름") or "")
out[_flat(name)] = name
for old in row.get("옛이름") or []:
out.setdefault(_flat(old), name)
return out
def trade_names() -> dict[str, str]:
return _trade_names(id(m01.loaded()[1]))
def sources(text: str) -> list[list[str]]:
"""글 → [[책, 절 번호]] — 「건설공사표준품셈 3-2-1」 · 「산림사업표준품셈(9-14)」."""
out = []
for hit in _BOOK.finditer(text):
num = labor_inputs._NUMBER.search(text, hit.end())
if not num or num.start() - hit.end() > 20:
continue
before = text[max(0, hit.start() - 12) : hit.start()]
book = "산림품셈" if "산림" in before or "임도" in before else "건설품셈"
if [book, num.group()] not in out:
out.append([book, num.group()])
return out
def _columns(rows: list[list[Any]]) -> tuple[int, int, dict[str, int]]:
"""(자료 첫 줄, 산출근거 열, {계 · 노무비 · 재료비 · 경비: 열}) — 머리 없으면 (0, 1, {})."""
for at, row in enumerate(rows[:10]):
flat = [_flat(v) for v in row]
if "산출근거" in flat:
money = {key: flat.index(head) for head, key in _MONEY.items() if head in flat}
return at + 1, flat.index("산출근거"), money
return 0, 1, {}
def read(rows: list[list[Any]] | None) -> dict[int, dict[str, Any]]:
"""단가산출 시트 줄 → `{N: {번, 이름, 단위, 행: [시작, 끝], 출처, 직종, 금액}}`
(행 = 엑셀 1부터)."""
if not rows:
return {}
start, col, money = _columns(rows)
names = trade_names()
blocks: dict[int, dict[str, Any]] = {}
block: dict[str, Any] | None = None
for at, row in enumerate(rows[start:], start + 1):
text = row[col] if col < len(row) else None
if not isinstance(text, str) or not text.strip():
continue
head = None if text[:1].isspace() else _HEAD.match(text)
if head:
block = blocks.setdefault(
int(head.group(1)),
{
"번": int(head.group(1)),
"이름": head.group(2),
"단위": head.group(3).strip(),
"행": [at, at],
"출처": [],
"직종": [],
"금액": {
key: v
for key, c in money.items()
for v in [row[c] if c < len(row) else None]
if isinstance(v, (int, float)) and not isinstance(v, bool)
},
},
)
continue
if block is None:
continue
block["행"][1] = at
block["출처"] += [one for one in sources(text) if one not in block["출처"]]
trade = _TRADE.match(text)
name = names.get(_flat(trade.group(1))) if trade else None
if name and name not in block["직종"]:
block["직종"].append(name)
return blocks
# ── 우리 쪽 조항 · 직종 ────────────────────────────────────────────────
@lru_cache(maxsize=2)
def _ours(whole_id: int) -> dict[str, dict[str, Any]]:
"""{로직 키: {책, 부문, 구분, 조항: {절 번호}, 직종: {이름}}} — 조항 = `sources()` 묶음 ·
직종 = 호표 인력(LB) 요소."""
files, whole = m01.loaded()
lb = {str(row.get("키")): str(row.get("이름")) for row in whole.numbers.get("LB", {}).values()}
out = {}
for data in files.values():
if data.get("그룹") != "로직":
continue
for row in data.get("줄") or []:
kind = str(row.get("구분") or "")
book, _, part = kind.partition(" ")
trades = {
lb[str(h.get("요소"))]
for h in row.get("호표") or []
if isinstance(h, dict) and str(h.get("요소")) in lb
}
clauses = set(labor_inputs._NUMBER.findall(str(labor_inputs.sources(row) or "")))
out[str(row.get("키"))] = {
"책": book,
"부문": part,
"구분": kind,
"조항": clauses,
"직종": trades,
}
return out
def ours() -> dict[str, dict[str, Any]]:
return _ours(id(m01.loaded()[1]))
def _head(number: str) -> str:
return number.rsplit("-", 1)[0]
def _clause(theirs: list[list[str]], our: dict[str, Any]) -> tuple[float, list[str], str]:
"""(점수, 맞은 우리 조항 글, 꼬리표) — 같은 책만."""
nums = [n for book, n in theirs if book == our["책"]]
same = sorted(n for n in nums if n in our["조항"])
if same:
return 1.0, [f"{our['구분']} {n}" for n in same], "조항 같음"
heads = {_head(n) for n in nums}
near = sorted(n for n in our["조항"] if _head(n) in heads)
if near:
return 0.5, [f"{our['구분']} {n}" for n in near], "앞 마디 같음"
return 0.0, [], ""
def _text(line: dict[str, Any]) -> str:
text = f"{line.get('이름') or ''} {line.get('규격') or ''}"
for old, new in _TYPOS:
text = text.replace(old, new)
return text
def candidates(line: dict[str, Any], got: dict[str, Any]) -> list[dict[str, Any]]:
"""근거 후보 — 줄 `근거.출처` 가 없으면 빈 것 · `{로직, 이름, 단위, 점수, 근거}` 점수 차례."""
basis = line.get("근거") or {}
theirs = [list(one) for one in basis.get("출처") or []]
if not theirs:
return []
trades = set(basis.get("직종") or []) - {COMMON}
text = _text(line)
mine = hopyo_match.words(text) # 괄호 글 뺌 — 「잔토처리(인력)」 ↔ 「철골재 철거(인력)」
grams = hopyo_match._grams(hopyo_match.mm.plain(text)) # 괄호 글 포함 — 같은 점수 가름
table = ours()
scored = []
for one in got["로직"]:
our = table.get(one["로직"])
if not our:
continue
score, clauses, tag = _clause(theirs, our)
if not score:
continue
shared = sorted(trades & our["직종"])
if not shared and not any(a in b or b in a for a in mine for b in one["words"]):
continue
tags = [tag]
if our["책"] == "건설품셈" and our["부문"] in BONUS_PARTS:
score += 0.1
tags.append(f"건설 {our['부문']}")
if shared:
score += 0.5
tags.append("직종 " + " · ".join(shared))
unit = hopyo_match.fits(line.get("단위"), one["단위"])
if unit:
score += 0.15
tags.append("단위 맞음")
dice = 2 * len(grams & one["grams"]) / (len(grams) + len(one["grams"]))
why = {"조항": clauses, "직종": shared, "단위": unit, "점수": round(score, 3)}
scored.append(
(score, dice, {**hopyo_match._public(one), "근거": {**why, "까닭": " · ".join(tags)}})
)
scored.sort(key=lambda x: (-x[0], -x[1]))
return [one for _, _, one in scored]
def merge(basis: list[dict[str, Any]], named: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""근거 후보 + 이름 후보 → 로직 키로 합침(점수 = 이름 점수 + 근거 점수) · 점수 차례."""
if not basis:
return named
by_name = {c.get("로직"): c["점수"] for c in named}
out = [{**c, "점수": round(c["근거"]["점수"] + by_name.pop(c["로직"], 0), 3)} for c in basis]
out += [c for c in named if c.get("로직") in by_name]
return sorted(out, key=lambda c: -c["점수"])
+22 -4
View File
@@ -11,6 +11,9 @@
검산 = 줄 금액 floor(단가 × 수량 × 10) / 10 · 합계 단가 = 셋 합 ·
% 줄 = floor(바탕 × 율 / 100 × 10) / 10 · 호표 비목 = floor(줄 금액 합) · 총괄표 합계 —
어긋남은 글 목록(막지 않음).
행(329-1) = 호표 [제N호표 줄, 마지막 구성 줄] · 구성 줄 한 줄(엑셀 1부터) ·
근거(330-1) = 단산N참조 줄에 「단가산출」 블록 `{출처, 직종, 행}`(`M02_Hopyo_Basis.read`) ·
`단산` = 블록마다 `{번, 이름, 단위, 행}`.
"""
from __future__ import annotations
@@ -20,6 +23,8 @@ import re
from decimal import ROUND_FLOOR, Decimal, InvalidOperation
from typing import Any
from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis
SHEET = "일위대가"
SUMMARY = "일위대가총괄표"
COSTS = ("노무비", "재료비", "경비")
@@ -67,8 +72,8 @@ def _out(value: Decimal | None) -> int | float | None:
# ── 파일 → 시트 표(줄마다 값 목록) ────────────────────────────────────────
def _sheets(name: str, data: bytes) -> dict[str, list[list[Any]]]:
"""시트 이름 → 줄 목록 — 일위대가 · 총괄표만 읽음."""
wanted = (SHEET, SUMMARY)
"""시트 이름 → 줄 목록 — 일위대가 · 총괄표 · 단가산출만 읽음."""
wanted = (SHEET, SUMMARY, hopyo_basis.SHEET)
if not data:
raise ExcelError("빈 파일")
try:
@@ -212,7 +217,7 @@ def read_book(name: str, data: bytes) -> dict[str, Any]:
start, cols = _columns(rows, ("합계", *COSTS))
hopyos: list[dict[str, Any]] = []
raw: list[list[dict[str, Any]]] = []
for row in rows[start:]:
for at, row in enumerate(rows[start:], start + 1):
head = _flat(_cell(row, cols.get("호표")))
name_ = _text(_cell(row, cols.get("품명")))
hit = _HOPYO.fullmatch(head)
@@ -224,6 +229,7 @@ def read_book(name: str, data: bytes) -> dict[str, Any]:
"규격": _text(_cell(row, cols.get("규격"))),
"단위": _text(_cell(row, cols.get("단위"))),
"금액": _money(row, cols, 1),
"행": [at, at],
}
)
raw.append([])
@@ -244,9 +250,15 @@ def read_book(name: str, data: bytes) -> dict[str, Any]:
"참조": {"종류": ref.group(1), "번": int(ref.group(2))} if ref else None,
}
line["갈래"] = kind_of(line)
line["행"] = hopyos[-1]["행"][1] = at
raw[-1].append(line)
if not hopyos:
raise ExcelError(f"{SHEET} 시트에 호표(제N호표)가 없음")
blocks = hopyo_basis.read(sheets.get(hopyo_basis.SHEET))
for line in (one for lines in raw for one in lines):
ref = line["참조"] or {}
if ref.get("종류") == "단산" and (block := blocks.get(ref["번"])):
line["근거"] = {k: block[k] for k in ("출처", "직종", "행")}
wrong: list[str] = []
totals = _summary(sheets.get(SUMMARY))
for hopyo, lines in zip(hopyos, raw, strict=True):
@@ -258,7 +270,13 @@ def read_book(name: str, data: bytes) -> dict[str, Any]:
{**_json(hopyo), "줄": [_json(line) for line in lines]}
for hopyo, lines in zip(hopyos, raw, strict=True)
]
return {"시트": SHEET, "호표": out, "검산": {"어긋남": wrong, "총괄표": len(totals)}}
basis = [{k: b[k] for k in ("번", "이름", "단위", "행")} for b in blocks.values()]
return {
"시트": SHEET,
"호표": out,
"단산": basis,
"검산": {"어긋남": wrong, "총괄표": len(totals)},
}
def _json(value: Any) -> Any:
+211
View File
@@ -0,0 +1,211 @@
"""M02 일위대가 실무자료 엑셀꼴 격자(PLAN 329-1) — 올린 엑셀 「일위대가」 · 「단가산출」 시트 →
A00 산출근거 스프레드시트 Workbook(`A00_Common/spreadsheet/spreadsheet_types.ts` 꼴 ·
읽기 전용 보기).
xls = xlrd `formatting_info` — 값 · 병합 · 열 폭(÷256 글자) ·
행 높이(÷20 pt · 시트 기본과 다른 것만) · 서식 표(글꼴 · 굵게 · 맞춤 · 줄바꿈 · 테두리 · 채움 ·
숫자 형식) · 틀고정(파일 값 · 없으면 머리 4행).
xlsx = 값만 · 넓은 시트는 머리 글(앞 4 줄) 마지막 열까지.
결과 `{책, 호표: [{번, 이름, 행}], 단산: [{번, 이름, 행}]}` — 행 = 엑셀 1부터 [시작, 끝].
"""
from __future__ import annotations
import io
from typing import Any
from openpyxl.utils import get_column_letter
from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis
from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel
SHEETS = (hopyo_excel.SHEET, hopyo_basis.SHEET)
HEAD_ROWS = 4
_HALIGN = {1: "left", 2: "center", 3: "right", 6: "centerContinuous", 7: "distributed"}
_VALIGN = {0: "top", 1: "center"}
_LINES = {1: "thin", 2: "medium", 3: "dashed", 4: "dotted", 5: "thick", 6: "double", 7: "hair"}
_SIDES = (("위", "top"), ("아래", "bottom"), ("왼", "left"), ("오른", "right"))
class _Styles:
"""서식 표 — 같은 서식은 한 번만 · [0] = 기본 `{}`."""
def __init__(self) -> None:
self.table: list[dict[str, Any]] = [{}]
self._index: dict[str, int] = {"{}": 0}
def add(self, style: dict[str, Any]) -> int:
key = repr(sorted(style.items()))
if not style:
return 0
if key not in self._index:
self._index[key] = len(self.table)
self.table.append(style)
return self._index[key]
def _rgb(book: Any, index: int) -> str | None:
got = book.colour_map.get(index)
return "#{:02X}{:02X}{:02X}".format(*got) if got else None
def _xf_style(book: Any, xf: Any, base: Any) -> dict[str, Any]:
font = book.font_list[xf.font_index]
out: dict[str, Any] = {}
if font.name != base.name:
out["글꼴"] = font.name
if font.height != base.height:
out["크기"] = font.height / 20
for key, on in (
("굵게", font.bold),
("기울임", font.italic),
("밑줄", font.underline_type),
("취소선", font.struck_out),
):
if on:
out[key] = True
color = _rgb(book, font.colour_index)
if color and color != "#000000":
out["글자색"] = color
if xf.background.fill_pattern == 1 and (fill := _rgb(book, xf.background.pattern_colour_index)):
out["채움"] = fill
align = xf.alignment
if align.hor_align in _HALIGN:
out["가로"] = _HALIGN[align.hor_align]
if align.vert_align in _VALIGN:
out["세로"] = _VALIGN[align.vert_align]
if align.text_wrapped:
out["줄바꿈"] = True
sides = {}
for name, side in _SIDES:
line = getattr(xf.border, f"{side}_line_style")
if line:
one = {"선": _LINES.get(line, "dashed")}
color = _rgb(book, getattr(xf.border, f"{side}_colour_index"))
if color and color != "#000000":
one["색"] = color
sides[name] = one
if sides:
out["테두리"] = sides
fmt = book.format_map.get(xf.format_key)
if fmt and fmt.format_str and fmt.format_str != "General":
out["형식"] = fmt.format_str
return out
def _value(cell: Any) -> Any:
import xlrd
if cell.ctype == xlrd.XL_CELL_TEXT:
return cell.value or None
if cell.ctype in (xlrd.XL_CELL_NUMBER, xlrd.XL_CELL_DATE):
return int(cell.value) if cell.value == int(cell.value) else cell.value
if cell.ctype == xlrd.XL_CELL_BOOLEAN:
return bool(cell.value)
if cell.ctype == xlrd.XL_CELL_ERROR:
return xlrd.error_text_from_code.get(cell.value, "#N/A")
return None
def _xls_sheet(book: Any, sh: Any, styles: _Styles, sid: str) -> dict[str, Any]:
base = book.font_list[0]
by_xf: dict[int, int] = {}
cells: dict[str, dict[str, Any]] = {}
for r in range(sh.nrows):
for c in range(sh.ncols):
cell = sh.cell(r, c)
if cell.xf_index not in by_xf:
by_xf[cell.xf_index] = styles.add(
_xf_style(book, book.xf_list[cell.xf_index], base)
)
one: dict[str, Any] = {}
if (value := _value(cell)) is not None:
one["값"] = value
if by_xf[cell.xf_index]:
one["서식"] = by_xf[cell.xf_index]
if one:
cells[f"{get_column_letter(c + 1)}{r + 1}"] = one
merged = [
f"{get_column_letter(c0 + 1)}{r0 + 1}:{get_column_letter(c1)}{r1}"
for r0, r1, c0, c1 in sh.merged_cells
]
cols = {
get_column_letter(c + 1): {
"폭": round(info.width / 256, 2),
**({"숨김": True} if info.hidden else {}),
}
for c, info in sh.colinfo_map.items()
if c < sh.ncols
}
rows = {
str(r + 1): {"높이": info.height / 20, **({"숨김": True} if info.hidden else {})}
for r, info in sh.rowinfo_map.items()
if r < sh.nrows and (info.height != sh.default_row_height or info.hidden)
}
frozen = sh.panes_are_frozen and (sh.horz_split_pos or sh.vert_split_pos)
freeze = {"행": sh.horz_split_pos, "열": sh.vert_split_pos} if frozen else {"행": HEAD_ROWS}
out = {
"id": sid,
"이름": sh.name.strip(),
"칸": cells,
"병합": merged,
"열": cols,
"행": rows,
"틀고정": {k: v for k, v in freeze.items() if v},
}
if sh.default_row_height:
out["기본"] = {"행높이": sh.default_row_height / 20}
return out
def _xls(data: bytes) -> dict[str, Any]:
import xlrd
book = xlrd.open_workbook(file_contents=data, formatting_info=True, logfile=io.StringIO())
styles = _Styles()
found = {sh.name.strip(): sh for sh in book.sheets()}
sheets = [
_xls_sheet(book, found[name], styles, f"s{at}")
for at, name in enumerate((n for n in SHEETS if n in found), 1)
]
font = book.font_list[0]
return {
"기본": {"글꼴": font.name, "크기": font.height / 20},
"서식": styles.table,
"시트": sheets,
}
def _xlsx(name: str, data: bytes) -> dict[str, Any]:
# ponytail: xlsx 는 값만(병합 · 폭 · 서식은 openpyxl 비 read_only 로 무거움) — 필요하면 그때
found = hopyo_excel._sheets(name, data)
sheets = []
for at, title in enumerate((n for n in SHEETS if n in found), 1):
rows = found[title]
width = max(
(c + 1 for row in rows[:HEAD_ROWS] for c, v in enumerate(row) if v not in (None, "")),
default=0,
)
cells = {
f"{get_column_letter(c + 1)}{r + 1}": {
"값": v if isinstance(v, (int, float, bool)) else str(v)
}
for r, row in enumerate(rows)
for c, v in enumerate(row[:width])
if v not in (None, "")
}
sheets.append({"id": f"s{at}", "이름": title, "칸": cells, "틀고정": {"행": HEAD_ROWS}})
return {"서식": [{}], "시트": sheets}
def read(name: str, data: bytes, parsed: dict[str, Any]) -> dict[str, Any]:
"""엑셀 → `{책: Workbook, 호표, 단산}` — `parsed` = `M02_Hopyo_Excel.read_book` 결과
(행 범위)."""
xlsx = data[:4] == b"PK\x03\x04" or name.lower().endswith(".xlsx")
book = _xlsx(name, data) if xlsx else _xls(data)
keep = ("번", "이름", "행")
return {
"책": {"종류": "통합문서", "판": 1, "열": "", **book, "활성": "s1"},
"호표": [{k: h.get(k) for k in keep} for h in parsed.get("호표") or []],
"단산": [{k: b.get(k) for k in keep} for b in parsed.get("단산") or []],
}
+11 -8
View File
@@ -4,6 +4,8 @@
씨앗(일위대가조합.json 메모 · 읽기만)
③ 정리 이름 같음(괄호 · 기계 낱말) ④ 후보 5(글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 ·
뜻 낱말이 하나도 안 겹치면 뺌 — 「잔토처리」 ↔ 「소금처리」) —
+ 근거 후보(330-2 · `M02_Hopyo_Basis` · 단산 줄 출처 · 직종 · 이름 거름 안 거침 ·
로직 키로 합쳐 점수 더함) ·
③ ④ 는 사용자 확정(층 기억) 뒤에만 들어감.
줄 갈래: 단가산출 · 기계 · 기타 = 로직 / 자재 · 별도계상 = 자재(M01 자재품목) / 호표 = 그 호표 /
% = 안 들어감(로직 덧줄이 셈).
@@ -27,6 +29,7 @@ from typing import Any
from M01_MasterData import M01_MasterData_Store as m01
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_Labor_Inputs as labor_inputs
from M02_MasterTemplete import M02_Labor_Ref as refs
@@ -295,9 +298,10 @@ def _exact(line: dict[str, Any], pool: str, got: dict[str, Any]) -> list[dict[st
def candidates(
line: dict[str, Any], pool: str, got: dict[str, Any] | None = None, limit: int = 5
line: dict[str, Any], pool: str, got: dict[str, Any] | None = None, limit: int | None = 5
) -> list[dict[str, Any]]:
"""③ 정리 이름 같음 먼저 · ④ 글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 — 위 `limit`."""
"""③ 정리 이름 같음 먼저 · ④ 글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 — 위 `limit`
(None = 모두)."""
me = tidy(line.get("이름"))
grams, unit = _grams(me), line.get("단위")
text = mm.plain_spec(f"{line.get('이름')}{line.get('규격')}")
@@ -402,11 +406,9 @@ def match_line(
return out
top = [{**_public(x), "점수": 9} for x in found]
same = [_public(x) for x in found]
rest = [
c
for c in candidates(line, pool, got)
if {k: v for k, v in c.items() if k != "점수"} not in same
]
basis = hopyo_basis.candidates(line, got) if pool == "로직" else []
named = candidates(line, pool, got, None if basis else 5)
rest = [c for c in hopyo_basis.merge(basis, named) if _public(c) not in same]
out["후보"] = (top + rest)[:5]
return {**out, "상태": "후보" if out["후보"] else "없음"}
@@ -461,7 +463,8 @@ def view(folder: Path, key: str, no: int | None = None) -> dict[str, Any]:
if no is None:
got = targets()
hopyos = [match_hopyo(h, memo, units, False, got) for h in doc.get("호표") or []]
brief = [{k: h[k] for k in ("번", "이름", "규격", "단위", "금액", "맞춤")} for h in hopyos]
keep = ("번", "이름", "규격", "단위", "금액", "맞춤")
brief = [{k: h.get(k) for k in keep} for h in hopyos]
ready = [sum(h["맞춤"][0] for h in hopyos), sum(h["맞춤"][1] for h in hopyos)]
return {**hopyo_store.brief(doc), "검산": doc.get("검산"), "호표": brief, "맞춤": ready}
hopyo = next((h for h in doc.get("호표") or [] if int(h["번"]) == int(no)), None)
+27 -3
View File
@@ -1,5 +1,5 @@
"""M02 일위대가 실무자료 저장(PLAN 324-1) — 층 폴더 `실무자료/<키>.json` ·
맞춤 기억 `실무자료/맞춤.json`.
맞춤 기억 `실무자료/맞춤.json` · 엑셀꼴 격자 `실무자료/격자/<키>.json`(329-1 · 파일과 같이 지움).
파일 = 읽은 결과만(원본 엑셀 안 둠) `{키, 파일, 올린때, 올린이, 시트, 호표, 검산}` · 키 `HP-######`
(층마다 다음 번호) · 맞춤 기억 = `{줄: {줄 열쇠: {로직?|자재?|빼기?, 입력값, 환산, 확정}}}`.
@@ -20,6 +20,8 @@ from M02_MasterTemplete import M02_MasterTemplete_Store as store
DIRNAME = "실무자료"
MEMORY = "맞춤"
GRID = "격자"
NO_GRID = "격자 없음 — 다시 올리면 보임"
_KEY = re.compile(r"^HP-(\d{6})$")
# ponytail: 프로세스 안 잠금 — 서버를 여럿 띄우면 파일 잠금으로
_LOCK = threading.Lock()
@@ -42,6 +44,10 @@ def _load(path: Path) -> Any:
return None
def _grid_path(folder: str | Path, key: str) -> Path:
return _dir(folder) / GRID / _path(folder, key).name
def brief(doc: dict[str, Any]) -> dict[str, Any]:
"""목록 한 줄 — `{키, 파일, 올린때, 올린이, 시트, 호표수, 어긋남}`."""
keep = ("키", "파일", "올린때", "올린이", "시트")
@@ -69,8 +75,14 @@ def read(folder: str | Path, key: str) -> dict[str, Any]:
return doc
def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dict[str, Any]:
"""읽은 결과 → 새 키로 씀 → 목록 한 줄."""
def save(
folder: str | Path,
name: str,
parsed: dict[str, Any],
who: Any,
grid: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""읽은 결과(+ 격자) → 새 키로 씀 → 목록 한 줄."""
with _LOCK:
have = [
int(m.group(1)) for p in _dir(folder).glob("HP-*.json") if (m := _KEY.match(p.stem))
@@ -83,16 +95,28 @@ def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dic
"올린이": who,
**parsed,
}
if grid is not None:
book = {**grid["책"], "열": key}
atomic_write_json(_grid_path(folder, key), {**grid, "책": book})
atomic_write_json(_path(folder, key), doc)
return brief(doc)
def grid(folder: str | Path, key: str) -> dict[str, Any]:
"""엑셀꼴 격자 `{책, 호표, 단산}` — 없으면(옛 파일) 404."""
doc = _load(_grid_path(folder, key))
if not isinstance(doc, dict):
raise store.StoreError(404, NO_GRID)
return doc
def delete(folder: str | Path, key: str) -> None:
path = _path(folder, key)
with _LOCK:
if not path.is_file():
raise store.StoreError(404, f"없는 실무자료 「{key}」")
path.unlink()
_grid_path(folder, key).unlink(missing_ok=True)
# ── 맞춤 기억 ─────────────────────────────────────────────────────────
+17 -2
View File
@@ -20,6 +20,7 @@ from pydantic import BaseModel
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_Hopyo_Build as hopyo_build
from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel
from M02_MasterTemplete import M02_Hopyo_Grid as hopyo_grid
from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_Labor_Doc as labor
@@ -176,7 +177,11 @@ def _upload(name: str, data: bytes, who: dict[str, Any]) -> dict[str, Any]:
parsed = hopyo_excel.read_book(name, data)
except hopyo_excel.ExcelError as error:
raise store.StoreError(400, str(error)) from error
return hopyo_store.save(store.root(), name, parsed, who)
try:
grid = hopyo_grid.read(name, data, parsed)
except Exception: # 격자는 보기용 덤 — 못 만들어도 올리기는 됨(보기 창 404)
grid = None
return hopyo_store.save(store.root(), name, parsed, who, grid)
@router.get("/layers/{layer}/hopyo")
@@ -205,7 +210,7 @@ def hopyo_upload(
owner: str | None = None,
session: Session = Depends(verify_session),
) -> dict:
"""엑셀(xls · xlsx) 올리기 — 「일위대가」 시트를 읽어 결과만 저장 · 20 MB ·
"""엑셀(xls · xlsx) 올리기 — 「일위대가」 시트를 읽어 결과 · 격자 저장 · 20 MB ·
엉터리 · 시트 없음 400."""
folder = _root(session, layer, owner, write=True)
data = file.file.read(HOPYO_MAX_BYTES + 1)
@@ -225,6 +230,16 @@ def hopyo_read(
return _call(folder, hopyo_match.view, folder, key)
@router.get("/layers/{layer}/hopyo-grid/{key}")
def hopyo_grid_read(
layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session)
) -> dict:
"""엑셀꼴 격자(329-1) `{책: Workbook, 호표: [{번, 이름, 행}], 단산: [{번, 이름, 행}]}` ·
옛 파일 404."""
folder = _root(session, layer, owner)
return _call(folder, hopyo_store.grid, folder, key)
@router.get("/layers/{layer}/hopyo/{key}/{no}")
def hopyo_one(
layer: Layer,
@@ -0,0 +1,231 @@
"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3) — `B05_Profile_Engine_RouteInitial_Store`.
./venv/Scripts/python.exe -m pytest resources/tester/test_224_3_recommend_store.py -q
같은 입력 = 같은 키 · 하나만 바뀌어도 다른 키 · 두 번째 요청 셈 0회 · 덮어쓰기 없음 ·
[초기화](스냅샷 복원) 뒤에도 남음 · 자료 교체 때 지워짐 · 덮개 수는 꺼낼 때 지금 관으로.
엔드포인트 몸통(`compute_initial`)은 DB · 지형 읽기만 갈아 끼우고 그대로 돌린다.
"""
import asyncio
import json
import shutil
from contextlib import asynccontextmanager
from pathlib import Path
from uuid import uuid4
import numpy as np
import pytest
from B05_Profile import B05_Profile_Router_Replan as replan
from B05_Profile import B05_Profile_Router_RouteInitial as router
from B05_Profile.B05_Profile_Engine_RouteInitial import STREAM_OFFSET_M, Terrain
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
DEFAULT_MODES,
request_fingerprint,
store_dir,
write_stored,
)
from common_util.common_util_initial_snapshot import (
_FILE_TREES,
restore_snapshot_files,
snapshot_dir,
)
from common_util.common_util_project_reset import purge_output_folders
def _terrain(lift: float = 0.0) -> Terrain:
x = np.arange(0.0, 40.0, 2.0)
y = np.arange(0.0, 20.0, 2.0)
xx, yy = np.meshgrid(x, y)
return Terrain(x, y, 0.1 * xx + lift, np.ones(xx.shape, dtype=bool))
EXPECTED = [(0.0, 10.0), (20.0, 10.0), (38.0, 10.0)]
STREAMS = [[(5.0, 0.0), (5.0, 18.0)]]
BASE = {
"mode": "weighted",
"weights": None,
"max_grade": 0.14,
"min_radius_m": 12.0,
"stream_offset_m": None,
"grade_smooth_m": None,
"target_grade": None,
"with_profile": True,
"design_speed": 20,
}
def _key(expected=EXPECTED, terrain=None, streams=STREAMS, **changes) -> str:
return request_fingerprint(expected, terrain or _terrain(), streams, **{**BASE, **changes})[0]
def test_same_input_same_key() -> None:
assert _key() == _key()
# 비운 값 = 기본값 · 비중 None = 50/25/25 — 화면이 채워 보내도 같은 키
assert _key() == _key(stream_offset_m=STREAM_OFFSET_M)
assert _key() == _key(weights={"expected": 50.0, "earthwork": 25.0, "grade": 25.0})
# 비중 갈래가 아니면 비중은 갈래가 정함 — 보낸 비중은 키에 안 듦
assert _key(mode="earthwork") == _key(mode="earthwork", weights={"expected": 1.0})
@pytest.mark.parametrize(
"change",
[
{"mode": "earthwork"},
{"weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}},
{"max_grade": 0.12},
{"min_radius_m": 15.0},
{"stream_offset_m": 50.0},
{"grade_smooth_m": 3.0},
{"target_grade": 0.08},
{"with_profile": False},
{"design_speed": 30},
{"expected": [(0.0, 10.0), (20.0, 11.0), (38.0, 10.0)]},
{"terrain": _terrain(lift=0.5)},
{"streams": []},
],
ids=lambda change: next(iter(change)),
)
def test_one_change_other_key(change: dict) -> None:
assert _key(**change) != _key()
def test_no_overwrite(tmp_path: Path) -> None:
assert write_stored(tmp_path, "k", "p", {}, {"n": 1}, True)
assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8"))
assert data["result"] == {"n": 1} and data["default"] is True
assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"}
def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
(tmp_path / "B05_Profile/route").mkdir(parents=True)
(tmp_path / "B05_Profile/route/initial_choice.json").write_text("{}", encoding="utf-8")
write_stored(tmp_path, "k", "p", {}, {}, True)
# 스냅샷 = 작업본 트리 사본(`save_initial_snapshot` 과 같은 자리)
for tree in _FILE_TREES:
source = tmp_path / tree
if source.is_dir():
shutil.copytree(source, snapshot_dir(tmp_path) / tree.replace("/", "__"))
restore_snapshot_files(tmp_path)
assert (store_dir(tmp_path) / "k.json").is_file()
purge_output_folders(tmp_path) # 자료 교체
assert not store_dir(tmp_path).exists()
class _Calls:
def __init__(self) -> None:
self.count = 0
@pytest.fixture
def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
"""`compute_initial` 의 DB · 파일 읽기를 갈아 끼움 · 셈은 횟수만 세는 가짜."""
import B05_Profile.B05_Profile_Engine_Grade as grade
import B05_Profile.B05_Profile_Engine_RouteInitial as engine
import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
import common_util.common_util_surface_confirmation as surface
calls = _Calls()
async def project_paths(_pid):
return tmp_path, "x"
async def plan_criteria(_pid):
return 12.0, 12.0, 5.0
async def road_settings(_pid):
return "간선", 20, "일반"
async def surface_params(_connection, _pid):
return {}
class _Pool:
@asynccontextmanager
async def acquire(self):
yield None
def generate(expected, terrain, streams, *, pipes, **_inputs):
assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김
calls.count += 1
return {
"planned": [[x, y] for x, y in expected],
"metrics": {
"length_m": 38.0,
"offset_avg_m": 0.0,
"profile": {
"stations_m": [0.0, 38.0],
"plan_m": [1.0, 1.0],
"ground_m": [0.0, 0.0],
},
},
}
monkeypatch.setattr(replan, "_project_paths", project_paths)
monkeypatch.setattr(replan, "_plan_criteria", plan_criteria)
monkeypatch.setattr(replan, "_road_settings", road_settings)
monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
monkeypatch.setattr(engine, "load_terrain_and_streams", lambda *_a: (_terrain(), STREAMS))
monkeypatch.setattr(engine, "generate_initial_route", generate)
# 덮개 여유 2 m 관 하나가 노선 위 — 계획고 − 지반고 1 m 라 모자람 1
monkeypatch.setattr(profile, "project_pipes", lambda _root: [(19.0, 10.0, 2.0)])
calls.root = tmp_path
return calls
def _ask(**fields) -> dict:
request = replan.RouteInitialRequest(**fields)
return asyncio.run(router.compute_initial(uuid4(), request))
def test_second_request_computes_nothing(wired: _Calls) -> None:
first = _ask(mode="earthwork")
assert wired.count == 1 and first["stored"] is False
second = _ask(mode="earthwork")
assert wired.count == 1 and second["stored"] is True
assert second["planned"] == first["planned"]
# 덮개 수는 꺼낼 때마다 지금 관으로 · 저장본에는 없음
assert second["metrics"]["profile"]["cover_short"] == 1
files = list(store_dir(wired.root).glob("*.json"))
assert len(files) == 1
saved = json.loads(files[0].read_text(encoding="utf-8"))
assert "cover_short" not in saved["result"]["metrics"]["profile"]
assert saved["default"] is True and saved["request"]["mode"] == "earthwork"
# 제약 하나 바꾸면 파일 하나 더 · 옛 파일 그대로 · 기본값 표시 꺼짐
changed = _ask(mode="earthwork", criteria={"min_radius_m": 15.0})
assert changed["stored"] is False and wired.count == 2
assert len(list(store_dir(wired.root).glob("*.json"))) == 2
assert json.loads(files[0].read_text(encoding="utf-8")) == saved
def test_same_pool_in_flight_waits(wired: _Calls) -> None:
async def both() -> list[dict]:
request = replan.RouteInitialRequest(mode="gentle")
project = uuid4()
return await asyncio.gather(
router.compute_initial(project, request), router.compute_initial(project, request)
)
results = asyncio.run(both())
assert wired.count == 1
assert sorted(result["stored"] for result in results) == [False, True]
def test_precompute_defaults_then_hits(wired: _Calls) -> None:
asyncio.run(router.precompute_defaults(uuid4()))
assert wired.count == len(DEFAULT_MODES) == 5
saved = [
json.loads(path.read_text(encoding="utf-8"))
for path in store_dir(wired.root).glob("*.json")
]
assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
assert all(item["default"] for item in saved)
# 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본
hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
follow = _ask(mode="follow", with_profile=False)
assert hit["stored"] and follow["stored"] and wired.count == 5
+108
View File
@@ -0,0 +1,108 @@
"""일위대가 실무자료 단가산출 근거 맞춤(PLAN 330 · `M02_Hopyo_Basis` · 맞춤 후보 `근거`).
블록 북부청 95 · 사방 67 · 단산93 근거 [건설 3-2-1] · [특별인부] ·
후보 93 → GC000083 · 21 → GC000198 · 64 → GC000143 위 둘 · 49 → GC000085 ·
잔토처리에 철골재 철거 없음 · 근거 없는 줄 그대로 · 상태 「후보」(자동 확정 없음) ·
master_data 그대로.
"""
import pytest
from M02_MasterTemplete import M02_Hopyo_Basis as hb
from M02_MasterTemplete import M02_Hopyo_Excel as hx
from M02_MasterTemplete import M02_Hopyo_Match as hm
from resources.tester import test_m02_hopyo_excel as base
from resources.tester.test_m02_hopyo_excel import (
NORTH,
SABANG,
URL,
_post,
master_hash,
)
client = base.client # 공용 고정물(임시 층)
@pytest.fixture(scope="module")
def north() -> dict:
return hx.read_book(NORTH.name, NORTH.read_bytes())
def _line(book: dict, no: int) -> dict:
"""단산 no 를 가리키는 첫 줄."""
return next(x for h in book["호표"] for x in h["줄"] if x["참조"] == {"종류": "단산", "번": no})
def test_블록_수_와_근거(north):
assert len(north["단산"]) == 95
assert [b["번"] for b in north["단산"]] == list(range(1, 96))
sabang = hx.read_book(SABANG.name, SABANG.read_bytes())
assert len(sabang["단산"]) == 67
assert _line(north, 93)["근거"] == {
"출처": [["건설품셈", "3-2-1"]],
"직종": ["특별인부"],
"행": [4229, 4244],
}
assert _line(north, 21)["근거"]["직종"] == ["콘크리트공", "보통인부"] # 「보 통 인부」 빈칸 뺌
assert _line(north, 64)["근거"]["출처"][0] == ["건설품셈", "4-1-1"]
lines = [x for h in north["호표"] for x in h["줄"] if x["갈래"] == "단가산출"]
assert all("근거" in x for x in lines) # 268 줄 모두 블록 있음
def test_출처_글():
assert hb.sources("[건설공사표준품셈 3-2-1 인력굴착]") == [["건설품셈", "3-2-1"]]
assert hb.sources("건설표줌품셈(3-1-3) 자갈") == [["건설품셈", "3-1-3"]]
assert hb.sources("[2025년 산림사업표준품셈 9-14]") == [["산림품셈", "9-14"]]
assert hb.sources("흙운반 [임도표준품셈 4-2]") == [["산림품셈", "4-2"]]
assert hb.sources("[건설표준품셈 준용]") == []
def _top(book: dict, no: int) -> list[dict]:
got = hm.match_line(_line(book, no), [], hm.units_of(book))
assert got["상태"] == "후보" and got["들어감"] is False # 자동 확정 없음
return got["후보"]
def test_근거_후보(north):
before = master_hash()
dig = _top(north, 93)
assert dig[0]["로직"] == "GC000083"
assert dig[0]["근거"] == {
"조항": ["건설품셈 공통 3-2-1"],
"직종": ["특별인부"],
"단위": True,
"점수": 1.75,
"까닭": "조항 같음 · 건설 공통 · 직종 특별인부 · 단위 맞음",
}
assert _top(north, 21)[0]["로직"] == "GC000198"
assert "GC000143" in [c["로직"] for c in _top(north, 64)[:2]]
assert _top(north, 49)[0]["로직"] == "GC000085"
soil = next(x for x in north["호표"][2]["줄"] if x["이름"].startswith("잔토처리"))
names = [c["이름"] for c in hm.match_line(soil, [], {})["후보"]]
assert not any("철골재" in n for n in names)
assert master_hash() == before
def test_근거_없는_줄_그대로(north):
bare = [
x
for h in north["호표"]
for x in h["줄"]
if x["갈래"] == "단가산출" and not x["근거"]["출처"]
]
assert bare
for line in bare:
plain = {k: v for k, v in line.items() if k != "근거"}
assert hm.match_line(line, [], {}) == hm.match_line(plain, [], {})
def test_길_응답_근거_칸(client):
before = master_hash()
assert _post(client).status_code == 200
res = client.get(f"{URL.format('personal')}/HP-000001/3")
lines = {x["이름"]: x for x in res.json()["호표"]["줄"]}
dig = lines["구조물터파기(인력100%)"]
assert dig["근거"]["출처"] == [["건설품셈", "3-2-1"]]
assert dig["맞춤"]["후보"][0]["근거"]["까닭"].startswith("조항 같음")
assert "근거" not in lines["임도표지석(받침,글새김포함)"] # 자재 줄
assert master_hash() == before
+120
View File
@@ -0,0 +1,120 @@
"""일위대가 실무자료 엑셀꼴 격자(PLAN 329-1 · `M02_Hopyo_Grid` · `GET hopyo-grid/{키}`).
북부청 임도 xls — 병합 11 · 15열 · B20 「제3호표」 · `#,##0` · 호표 68 행 범위 안 겹침 ·
제3호표 [20, 24] · 올리면 격자도 · 200 / 옛 파일 404 · 지우면 격자도 · 역할 · xlsx 값만 ·
master_data 그대로.
"""
import pytest
from M02_MasterTemplete import M02_Hopyo_Excel as hx
from M02_MasterTemplete import M02_Hopyo_Grid as hg
from M02_MasterTemplete import M02_Hopyo_Store as hs
from M02_MasterTemplete import M02_Template_Layers as layers
from resources.tester import test_m02_hopyo_excel as base
from resources.tester.test_m02_hopyo_excel import (
NORTH,
URL,
_post,
_xlsx,
master_hash,
)
client = base.client # 공용 고정물(임시 층)
GRID = "/api/m02/layers/{}/hopyo-grid/{}"
@pytest.fixture(scope="module")
def north() -> dict:
data = NORTH.read_bytes()
return hg.read(NORTH.name, data, hx.read_book(NORTH.name, data))
def _sheet(grid: dict, name: str) -> dict:
return next(s for s in grid["책"]["시트"] if s["이름"] == name)
def test_격자_모양(north):
book = north["책"]
assert (book["종류"], book["판"], book["활성"]) == ("통합문서", 1, "s1")
assert [s["이름"] for s in book["시트"]] == ["일위대가", "단가산출"]
sheet = _sheet(north, "일위대가")
assert len(sheet["병합"]) == 11 and "B1:O2" in sheet["병합"]
assert len(sheet["열"]) == 15 and sheet["열"]["C"]["폭"] == round(7208 / 256, 2)
assert sheet["틀고정"] == {"행": 4}
assert sheet["칸"]["B20"]["값"] == "제3호표"
assert sheet["칸"]["H5"]["값"] == 1040052
styles = book["서식"]
assert styles[0] == {} and any(s.get("형식") == "#,##0" for s in styles)
assert any(s.get("테두리") for s in styles)
assert _sheet(north, "단가산출")["칸"]["B4"]["값"] == "1.유용토운반(무대)[㎥]"
def test_호표_단산_행_범위(north):
rows = [h["행"] for h in north["호표"]]
assert len(rows) == 68
assert all(a[1] < b[0] for a, b in zip(sorted(rows), sorted(rows)[1:], strict=False))
assert next(h for h in north["호표"] if h["번"] == 3) == {
"번": 3,
"이름": "임도표지석",
"행": [20, 24],
}
assert len(north["단산"]) == 95
assert next(b for b in north["단산"] if b["번"] == 93)["행"] == [4229, 4244]
def test_줄_행(north):
book = hx.read_book(NORTH.name, NORTH.read_bytes())
three = next(h for h in book["호표"] if h["번"] == 3)
assert three["행"] == [20, 24] and [x["행"] for x in three["줄"]] == [21, 22, 23, 24]
def test_올리기_격자_200_404_지우기(client):
before = master_hash()
assert _post(client).status_code == 200
res = client.get(GRID.format("personal", "HP-000001"))
assert res.status_code == 200, res.text
got = res.json()
assert got["책"]["열"] == "HP-000001" and len(got["호표"]) == 68
folder = layers.personal_dir(7, 42)
path = folder / hs.DIRNAME / hs.GRID / "HP-000001.json"
assert path.is_file()
assert [r["키"] for r in client.get(URL.format("personal")).json()["줄"]] == ["HP-000001"]
path.unlink() # 옛 파일(324) = 격자 없음
old = client.get(GRID.format("personal", "HP-000001"))
assert old.status_code == 404 and old.json()["detail"] == "격자 없음 — 다시 올리면 보임"
assert _post(client).status_code == 200
assert client.get(GRID.format("personal", "HP-000002")).status_code == 200
assert client.delete(URL.format("personal") + "/HP-000002").status_code == 200
assert not (folder / hs.DIRNAME / hs.GRID / "HP-000002.json").exists()
assert client.get(GRID.format("personal", "HP-000002")).status_code == 404
assert client.get(GRID.format("personal", "../x")).status_code == 404
assert client.get(URL.format("personal") + "/HP-000001/3").status_code == 200 # 길 안 부딪힘
assert master_hash() == before
@pytest.mark.parametrize(
("role", "layer", "owner", "want"),
[
({"role": "USER"}, "personal", None, 200),
({"role": "USER"}, "personal", "7/43", 403),
({"role": "USER"}, "company", None, 200),
({"role": "USER"}, "system", None, 200),
],
)
def test_역할_읽기(client, role, layer, owner, want):
client.who.update({"role": "SYSTEM_ADMIN"})
assert _post(client, layer if want == 200 else "personal").status_code == 200
client.who.update(role)
params = {"owner": owner} if owner else {}
assert client.get(GRID.format(layer, "HP-000001"), params=params).status_code == want
def test_xlsx_값만():
data = _xlsx()
grid = hg.read("a.xlsx", data, hx.read_book("a.xlsx", data))
sheet = grid["책"]["시트"][0]
assert sheet["이름"] == "일위대가" and sheet["틀고정"] == {"행": 4}
assert sheet["칸"]["B5"] == {"값": "제 1 호표"} and sheet["칸"]["E6"] == {"값": 0.5}
assert "K6" not in sheet["칸"] # 머리 글(앞 4 줄) 마지막 열 J 까지
assert grid["호표"] == [{"번": 1, "이름": "표지석", "행": [5, 6]}] and grid["단산"] == []
+1 -1
View File
@@ -105,7 +105,7 @@ def test_이름_같음_후보_호표_퍼센트(client):
assert all(c["규격"] == "25-21-120" for c in remicon["후보"][:3]) # 규격 같은 자재 먼저
dig = _hopyo(client, 3)["구조물터파기(인력100%)"]
assert dig["상태"] == "후보" and len(dig["후보"]) == 5
assert dig["후보"][0]["이름"].startswith("구조물터파기")
assert dig["후보"][0]["이름"] == "굴착(인력/토사)" # 330-2 근거(건설 3-2-1 · 특별인부) 1순위
ref = _hopyo(client, 45)["고화매트재료비"]
assert ref["상태"] == "호표" and ref["대상"] == {"호표": 46, "단위": "㎡"} and ref["들어감"]
pct = _hopyo(client, 46)["잡재료비"]