eomsangdonandClaude Opus 5.5
e9e949112c
- 새 파일 — Words(식 풀기 · 장비 갈래 · 조건 낱말) · Finger(블록 글 · STmate 자원 → 지문 · 기계 줄 · 직접 줄 호표) · Shares(절 몫 벡터) · Ours(마스터 판 열쇠 · 우리 지문 캐시 tmp/m02_hopyo · 엔진 셈 캐시 · 자재 값 찾기 건너뜀) · Option(옵션 공간 · 거꾸로 찾기 · 견줌) · Fit(손잡이 한 곳 · 순위 · 뒤 셈 · 맞춤셈 캐시) · Combine(조합 · 섞인 줄 몫 먼저) · Verdict(블록 셈 · 맞음 / 다른 점 / 후보) - Match — b_order(이름 · 근거 차례 · 같음 근거 검사 · 옛 판 글 · 조항 오타) · clause(윗마디 0.9 · 나 창 Basis.clause 있으면 그것) · 이름 후보 로직은 접지 않음 - 정답 표 fixtures/m02_hopyo_golden.json(사람 136 · 새 파일 51) · 골든 도우미 - 북부청 92 줄 실측 — 1순위 87 · 다섯 안 91 · 맞음 41 틀림 0 · 다른 점 28 틀림 0 · 조합 맞음 57/66 · 우리 지문 셈 37 초 · 캐시 읽기 0.05 초 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD
365 lines
15 KiB
Python
365 lines
15 KiB
Python
"""M02 일위대가 맞춤 — 우리 돈 로직 지문 · 마스터 판 열쇠 · 디스크 캐시 · 엔진 셈 캐시
|
|
(PLAN 338-1 · 설계 2-3).
|
|
|
|
지문 = 정적(호표 인력 → 직종 · 「로직(기계경비 …)」 → 장비 갈래 · 규격 · 재료 이름 · 덧줄 % ·
|
|
이름 · 고르기 글 조건 낱말 · 기계경비 로직) + 변형(고르기 입력 곱 64 이하 모두 · 넘으면 하나씩 ·
|
|
지역 · 조종원직종 · 보정작업 · 기종 목록 칸 뺌)마다 단위당 L(직종 인) · X(장비 시간) · M(자재) ·
|
|
S(딸린 로직) · mix(인력비율 · 기계비율).
|
|
캐시 = `tmp/m02_hopyo/우리지문_<판>.json`(환경 `AISLO_HOPYO_CACHE` 로 바꿈) · 판 = 마스터 첫 층 JSON
|
|
sha256 묶음 · 처음 셈은 뒤 스레드(≈ 2 분) · 다음은 읽기만.
|
|
엔진 = `Labor_Doc.run_line` 과 같은 길(자동값 + 입력 → `mf.run`) · 셈 잠금 하나(뒤 셈 · 요청 겹침) ·
|
|
메모리 캐시. 원본(`resources/master_data`) 읽기만. 규칙은 336 실험(C4 `s2_ours_fp` · K5 `run`)
|
|
그대로.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import itertools
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_json import atomic_write_json
|
|
from M01_MasterData import M01_MasterData_Store as m01
|
|
from M01_MasterData import M01_MasterData_Store_Cache as m01_cache
|
|
from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match
|
|
from M02_MasterTemplete.M02_Hopyo_Words import conds, mtypes, specnums
|
|
|
|
FP_VERSION = 1 # 지문 꼴이 바뀌면 올림 — 옛 캐시 버림
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
_PER = re.compile(r"원\s*/\s*(\d[\d,]*\.?\d*|\.\d+)?")
|
|
_SUB = re.compile(r"^\s*로직\(\s*(\w+)\s*(?:,(.*))?\)\s*$")
|
|
EQ_CODE = re.compile(r"^\d{4}-\d{4}$")
|
|
_PCT = re.compile(r"\*\s*([\d.]+)\s*/\s*100")
|
|
SKIP_IN = ("지역", "조종원직종", "보정작업")
|
|
FIXED_MACH = {
|
|
"GC000267": {"굴착기"},
|
|
"GC000270": {"굴착기", "집게"},
|
|
"GC000271": {"굴착기", "브레이커"},
|
|
"GF000145": {"덤프트럭"},
|
|
}
|
|
RUN_CAP = 50_000 # ponytail: 엔진 셈 캐시 칸 수 — 넘으면 통째 비움(LRU 는 필요해지면)
|
|
_ENGINE = threading.RLock() # ponytail: 셈 잠금 하나 — 엔진을 뒤 · 요청 스레드가 같이 부를 때
|
|
_STATE: dict[str, Any] = {"판": None, "자국": None, "표": None, "스레드": None}
|
|
_RUN: dict[tuple, Any] = {}
|
|
_LOCK = threading.Lock()
|
|
_NO_PRICE = {str(i): "-" for i in range(400)} # 고른 자재 = 없는 키 → 값 없음 · 찾기 안 함
|
|
WHOLE_TTL = 2.0 # ponytail: 마스터 색인을 2 초 붙듦 — 줄마다 파일 자국을 다시 안 봄
|
|
|
|
|
|
def folder() -> Path:
|
|
return Path(os.environ.get("AISLO_HOPYO_CACHE") or ROOT / "tmp" / "m02_hopyo")
|
|
|
|
|
|
# ── 마스터 판 열쇠 ─────────────────────────────────────────────────────
|
|
def version() -> str:
|
|
"""마스터 판 = 첫 층 *.json(밑줄 뺌) 이름 + sha256 16 묶음의 sha256 앞 16 — 자국이 같으면 다시
|
|
안 셈."""
|
|
stamps = m01_cache._stamps(Path(m01.FOLDER))
|
|
if _STATE["자국"] != stamps:
|
|
parts = [
|
|
f"{name}:{m01.version_of((Path(m01.FOLDER) / name).read_bytes())}" for name in stamps
|
|
]
|
|
_STATE.update(자국=stamps, 판=hashlib.sha256("\n".join(parts).encode()).hexdigest()[:16])
|
|
return _STATE["판"]
|
|
|
|
|
|
def path_of(ver: str | None = None) -> Path:
|
|
return folder() / f"우리지문_{ver or version()}.json"
|
|
|
|
|
|
# ── 엔진 한 번 ─────────────────────────────────────────────────────────
|
|
def whole() -> Any:
|
|
now = time.monotonic()
|
|
if _STATE.get("색인") is None or now - _STATE.get("색인때", 0) > WHOLE_TTL:
|
|
_STATE.update(색인=m01.loaded()[1], 색인때=now)
|
|
return _STATE["색인"]
|
|
|
|
|
|
def is_mach_logic(row: dict[str, Any]) -> bool:
|
|
return "시간당" in str(row.get("이름") or "")
|
|
|
|
|
|
def eq_of(code: Any) -> tuple[set[str], set[float]]:
|
|
one = whole().numbers.get("EQ", {}).get(str(code)) or {}
|
|
if not one:
|
|
return set(), set()
|
|
return mtypes(f"{one.get('상세구분')} {one.get('이름')}"), specnums(one.get("규격"))
|
|
|
|
|
|
def times_of(row: dict[str, Any]) -> float:
|
|
per = _PER.findall(str(row.get("결과단위") or ""))
|
|
return float(per[0].replace(",", "")) if per and per[0] else 1.0
|
|
|
|
|
|
def _args(text: str | None) -> dict[str, str]:
|
|
out = {}
|
|
for part in re.split(r",(?![^()]*\))", text or ""):
|
|
if "=" in part:
|
|
k, v = part.split("=", 1)
|
|
out[k.strip()] = v.strip()
|
|
return out
|
|
|
|
|
|
def _labor_names() -> dict[str, str]:
|
|
w = whole()
|
|
if _STATE.get("LB판") is not w:
|
|
lb = {str(r.get("키")): str(r.get("이름")) for r in w.numbers.get("LB", {}).values()}
|
|
_STATE.update(LB판=w, LB=lb)
|
|
return _STATE["LB"]
|
|
|
|
|
|
def _engine(code: str, given: dict[str, Any]) -> dict[str, Any]:
|
|
"""{L, X, M, S, mix} — 줄 수량 ÷ 결과단위 수(단위당)."""
|
|
w = whole()
|
|
row = w.logic(code)
|
|
times = times_of(row)
|
|
by_name: dict[Any, dict] = {}
|
|
for h in row.get("호표") or []:
|
|
by_name.setdefault(h.get("이름"), h)
|
|
auto = {k: v for k, v in m01.mau.values(w, row).items() if v is not None}
|
|
with _ENGINE, m01.cm.mf.mp.choosing(_NO_PRICE): # 품만 씀 — 자재 값 찾기 건너뜀(수량 같음)
|
|
got = m01.cm.mf.run(w, code, {**auto, **m01._dec(given)})
|
|
mid, lb = got.get("중간") or {}, _labor_names()
|
|
L: dict[str, float] = {}
|
|
X: list[list] = []
|
|
M: dict[str, float] = {}
|
|
S: dict[str, float] = {}
|
|
for line in got.get("줄") or []:
|
|
h = by_name.get(line.get("이름"))
|
|
if h is None:
|
|
continue
|
|
try:
|
|
q = float(Decimal(str(line.get("수량") or 0))) / times
|
|
except ArithmeticError:
|
|
continue
|
|
kind = h.get("종류")
|
|
if kind == "인력":
|
|
t = lb.get(str(h.get("요소")), str(h.get("이름")))
|
|
L[t] = L.get(t, 0.0) + q
|
|
elif kind == "로직":
|
|
m = _SUB.match(str(h.get("요소") or ""))
|
|
sub = m.group(1) if m else ""
|
|
sub_row = w.logic(sub) if sub else {}
|
|
if sub_row and is_mach_logic(sub_row):
|
|
ts = set(FIXED_MACH.get(sub, set())) | mtypes(h.get("이름"))
|
|
sp = specnums(h.get("이름"))
|
|
for v in _args(m.group(2)).values():
|
|
lit = v.strip("'\"")
|
|
if not EQ_CODE.match(lit):
|
|
lit = str(mid.get(lit) or given.get(lit) or auto.get(lit) or "")
|
|
if EQ_CODE.match(lit):
|
|
a, b = eq_of(lit)
|
|
ts |= a
|
|
sp |= b
|
|
if q:
|
|
X.append([sorted(ts), sorted(sp), q])
|
|
elif q:
|
|
S[sub] = S.get(sub, 0.0) + q
|
|
elif kind == "재료" and q:
|
|
M[str(h.get("이름"))] = M.get(str(h.get("이름")), 0.0) + q
|
|
mix = {}
|
|
for k in ("인력비율", "기계비율"):
|
|
try:
|
|
if mid.get(k) not in (None, ""):
|
|
mix[k] = float(Decimal(str(mid.get(k))))
|
|
except ArithmeticError:
|
|
pass
|
|
return {"L": L, "X": X, "M": M, "S": S, "mix": mix}
|
|
|
|
|
|
def _mach_view(code: str, row: dict, given: dict, v: dict) -> dict:
|
|
"""「기계 시간당 기계경비」 = 고른 기계(EQ 칸)의 갈래 · 규격 1 시간(조종원은 기계경비 속)."""
|
|
auto = m01.mau.values(whole(), row)
|
|
X = []
|
|
for s in row.get("입력") or []:
|
|
opts = s.get("고르기") or []
|
|
if opts and all(isinstance(o, str) and EQ_CODE.match(o) for o in opts):
|
|
ts, sp = eq_of(given.get(s["이름"], auto.get(s["이름"])))
|
|
X.append([sorted(ts | FIXED_MACH.get(code, set())), sorted(sp), 1.0])
|
|
if not X and FIXED_MACH.get(code):
|
|
X = [[sorted(FIXED_MACH[code]), [], 1.0]]
|
|
return {**v, "L": {}, "X": X}
|
|
|
|
|
|
def run(code: str, given: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
"""엔진 한 번 → {L, X, M, S, mix} · 못 돌면 None(메모리 캐시)."""
|
|
given = dict(given or {})
|
|
key = (code, json.dumps(given, sort_keys=True, default=str, ensure_ascii=False))
|
|
if key not in _RUN:
|
|
try:
|
|
v = _engine(code, given)
|
|
row = whole().logic(code)
|
|
v = _mach_view(code, row, given, v) if is_mach_logic(row) else v
|
|
except Exception: # noqa: BLE001 — 못 도는 조합은 없음으로(엔진 예외가 여럿)
|
|
v = None
|
|
if len(_RUN) >= RUN_CAP:
|
|
_RUN.clear()
|
|
_RUN[key] = v
|
|
return _RUN[key]
|
|
|
|
|
|
# ── 지문 ───────────────────────────────────────────────────────────────
|
|
def _static(code: str, row: dict[str, Any], unit: str) -> dict[str, Any]:
|
|
picks = [s for s in row.get("입력") or [] if s.get("고르기")]
|
|
labels, opt_mach = [], []
|
|
for s in picks:
|
|
names = s.get("고르기이름") or {}
|
|
for v in s.get("고르기") or []:
|
|
if isinstance(v, str) and EQ_CODE.match(v):
|
|
ts, sp = eq_of(v)
|
|
if ts:
|
|
opt_mach.append([sorted(ts), sorted(sp)])
|
|
else:
|
|
labels.append(str(names.get(v) or v))
|
|
lb = _labor_names()
|
|
trades, mach, mats = set(), [], []
|
|
for h in row.get("호표") or []:
|
|
kind = h.get("종류")
|
|
if kind == "인력":
|
|
trades.add(lb.get(str(h.get("요소")), str(h.get("이름"))))
|
|
elif kind == "로직":
|
|
m = _SUB.match(str(h.get("요소") or ""))
|
|
ts = set(FIXED_MACH.get(m.group(1) if m else "", set())) | mtypes(h.get("이름"))
|
|
for v in _args(m.group(2) if m else "").values():
|
|
if EQ_CODE.match(v.strip("'\"")):
|
|
ts |= eq_of(v.strip("'\""))[0]
|
|
if ts:
|
|
mach.append([sorted(ts), sorted(specnums(h.get("이름")))])
|
|
elif kind == "재료":
|
|
el = h.get("요소")
|
|
words = (
|
|
(el.get("검색어") or el.get("이름") or el.get("상세구분"))
|
|
if isinstance(el, dict)
|
|
else ""
|
|
)
|
|
mats.append(f"{h.get('이름')} {words or ''}".strip())
|
|
pct = [
|
|
[str(d.get("이름")), float(v)]
|
|
for d in row.get("덧줄") or []
|
|
for v in _PCT.findall(str(d.get("식") or ""))
|
|
]
|
|
if is_mach_logic(row):
|
|
mach.append([sorted(FIXED_MACH.get(code, set())), []])
|
|
pick_text = " ".join(labels + [str(s.get("이름")) for s in picks])
|
|
return {
|
|
"로직": code,
|
|
"이름": str(row.get("이름") or ""),
|
|
"단위": unit,
|
|
"직종": sorted(trades),
|
|
"장비": mach,
|
|
"장비고름": opt_mach,
|
|
"자재": mats,
|
|
"덧줄율": pct,
|
|
"이름조건": {k: sorted(v) for k, v in conds(row.get("이름")).items()},
|
|
"고름조건": {k: sorted(v) for k, v in conds(pick_text).items()},
|
|
"기계경비": is_mach_logic(row),
|
|
}
|
|
|
|
|
|
def combos(row: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""변형 입력 — 고르기 곱 64 이하 모두 · 넘으면 하나씩(기본 + 칸마다)."""
|
|
picks = []
|
|
for s in row.get("입력") or []:
|
|
opts = s.get("고르기") or []
|
|
if not opts or any(w in str(s.get("이름")) for w in SKIP_IN):
|
|
continue
|
|
if any(isinstance(v, str) and EQ_CODE.match(v) for v in opts):
|
|
continue
|
|
picks.append((str(s["이름"]), list(opts)))
|
|
if not picks:
|
|
return [{}]
|
|
if math.prod(len(o) for _, o in picks) <= 64:
|
|
names = [n for n, _ in picks]
|
|
return [dict(zip(names, vals)) for vals in itertools.product(*[o for _, o in picks])]
|
|
return [{}] + [{n: v} for n, opts in picks for v in opts]
|
|
|
|
|
|
def _sig(v: dict) -> str:
|
|
r = lambda x: float(f"{x:.4g}") # noqa: E731
|
|
return json.dumps(
|
|
[
|
|
sorted((k, r(q)) for k, q in v["L"].items()),
|
|
sorted((tuple(a), tuple(b), r(q)) for a, b, q in v["X"]),
|
|
sorted((k, r(q)) for k, q in v["M"].items()),
|
|
],
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
def fingerprint(code: str, unit: str) -> dict[str, Any]:
|
|
"""로직 하나 → 정적 지문 + 변형(서로 다른 품 꼴만)."""
|
|
row = whole().logic(code)
|
|
fp = _static(code, row, unit)
|
|
seen, variants, errors = set(), [], 0
|
|
for given in combos(row) if not fp["기계경비"] else [{}]:
|
|
try:
|
|
v = _engine(code, given) # 기계경비 로직도 엔진 그대로(맞춤이 고르기 장비로 봄)
|
|
except Exception: # noqa: BLE001 — 못 도는 조합은 셈만
|
|
errors += 1
|
|
continue
|
|
if (k := _sig(v)) not in seen:
|
|
seen.add(k)
|
|
variants.append({"in": {n: str(x) for n, x in given.items()}, **v})
|
|
return {**fp, "갈래수": len(variants), "오류": errors, "변형": variants}
|
|
|
|
|
|
# ── 표 · 캐시 · 데움 ───────────────────────────────────────────────────
|
|
def _read(ver: str) -> list[dict] | None:
|
|
try:
|
|
doc = json.loads(path_of(ver).read_bytes())
|
|
except (OSError, ValueError):
|
|
return None
|
|
ok = isinstance(doc, dict) and doc.get("판") == ver and doc.get("지문판") == FP_VERSION
|
|
return doc.get("표") if ok else None
|
|
|
|
|
|
def build(ver: str | None = None) -> list[dict]:
|
|
"""1,438 돈 로직 지문을 셈 → 디스크 캐시(같은 판이면 덮음)."""
|
|
ver = ver or version()
|
|
table = [fingerprint(one["로직"], one["단위"]) for one in hopyo_match.targets()["로직"]]
|
|
atomic_write_json(path_of(ver), {"판": ver, "지문판": FP_VERSION, "표": table})
|
|
return table
|
|
|
|
|
|
def table() -> list[dict] | None:
|
|
"""우리 지문(판이 같은 것) · 준비 전 None — 디스크에 있으면 읽어 둠."""
|
|
ver = version()
|
|
with _LOCK:
|
|
if _STATE.get("표판") != ver:
|
|
_STATE.update(표=_read(ver), 표판=ver)
|
|
return _STATE["표"]
|
|
|
|
|
|
def ready() -> bool:
|
|
return table() is not None
|
|
|
|
|
|
def warm(build_missing: bool = False) -> threading.Thread | None:
|
|
"""디스크 캐시 읽기(뒤) · `build_missing` 이면 없을 때 뒤 스레드로 셈 — 셈 중이면 그 스레드."""
|
|
running = _STATE.get("스레드")
|
|
if running is not None and running.is_alive():
|
|
return running
|
|
|
|
def work() -> None:
|
|
try:
|
|
if table() is None and build_missing:
|
|
ver = version()
|
|
got = build(ver)
|
|
with _LOCK:
|
|
_STATE.update(표=got, 표판=ver)
|
|
except Exception: # noqa: BLE001 — 마스터가 깨져도 서버를 막지 않음(보기에서 「준비 중」)
|
|
pass
|
|
|
|
thread = threading.Thread(target=work, name="m02-hopyo-ours", daemon=True)
|
|
_STATE["스레드"] = thread
|
|
thread.start()
|
|
return thread
|