eomsangdonandClaude Opus 5.5
7ae7515773
- 추천 저장본: 프로젝트 접근(화면 기본 갈래)은 같은 판 · 같은 입력이면 그 파일 소스 지문으로 키를 다시 내 찾음(read_stored_any) · 지문 견줌 · 다시 셈은 전처리 체인만 - 초기 종단 저장본: 접근은 코드 지문 빼고 입력만 견줌 · 파일에 코드 지문 칸 · 지문 칸 없는 옛 파일은 그대로 읽음 - 분석 지형 고르기 목록: 설계 지형 아닌 쪽은 DB 완료 모델도 있어야 둠(없으면 숨김) - 노선 [저장] 설계 지형 바꿈에 모델 없음(LookupError) = 되돌림 + 409 사유 한 줄 - 시험 test_309_3 둘 더함 · test_505_2_terrain_sheet 새로 · test_133_1 DB 모델 맞춤 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3).
|
|
|
|
같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면
|
|
셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 종류마다 최근 `KEEP_PER_KIND` 벌).
|
|
|
|
층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리
|
|
`B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라
|
|
[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
|
|
|
|
키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
|
|
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관
|
|
지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을
|
|
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 다른 판 · 지문 파일은 안 읽고 남기며 30일 지난
|
|
것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀). 프로젝트 접근(화면
|
|
요청)은 같은 판 · 같은 입력이면 다른 소스 지문 저장본도 꺼냄(`read_stored_any` · 505-1 덧) — 지문
|
|
견줌 · 다시 셈은 전처리 체인에서만.
|
|
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import functools
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, AsyncIterator, Callable
|
|
|
|
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
|
|
|
|
# 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) ·
|
|
# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) ·
|
|
# 8 = 횡단 토공 · 표준 횡단면 = 저장 패널값 · 엔진 소스 지문(420)
|
|
ALGO_VERSION = 8
|
|
# 엔진 소스 지문에 드는 파일(저장소 뿌리 기준) — 추천 결과를 낳는 코드
|
|
SOURCE_GLOBS = (
|
|
"B05_Profile/B05_Profile_Engine_RouteInitial*.py",
|
|
"B05_Profile/B05_Profile_Engine_Sections_Core.py",
|
|
"B06_Section/B06_Section_Engine_Design*.py",
|
|
"common_util/common_util_route_polyline.py",
|
|
)
|
|
STORE_DIR = "B05_Profile/route_recommend"
|
|
# 종류(갈래 · 지형) 마다 남길 벌 수 — 최근에 적은 것이 남음(꺼내기는 읽기만 · 505-1)
|
|
KEEP_PER_KIND = 2
|
|
# 다른 판 · 소스 지문 저장본을 이 날수 지나면 미리 셈 때 지움(420-2)
|
|
STALE_DAYS = 30
|
|
# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
|
|
DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
|
|
|
|
# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
|
|
_LOCKS: dict[str, asyncio.Lock] = {}
|
|
# 묶음 지문마다 그 묶음을 짤 때 표준 횡단면(`forget_pool_on_standard`)
|
|
_POOL_STANDARDS: dict[str, str] = {}
|
|
|
|
|
|
def store_dir(project_root: Path) -> Path:
|
|
return Path(project_root) / STORE_DIR
|
|
|
|
|
|
@functools.cache
|
|
def source_fingerprint() -> str:
|
|
"""엔진 소스 지문 — `SOURCE_GLOBS` 파일 내용(줄끝 LF 로 맞춤 · 두 PC 같은 코드 = 같은 지문)."""
|
|
root = Path(__file__).resolve().parents[1]
|
|
digest = hashlib.blake2b(digest_size=8)
|
|
for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}):
|
|
digest.update(path.name.encode())
|
|
digest.update(path.read_bytes().replace(b"\r\n", b"\n"))
|
|
return digest.hexdigest()
|
|
|
|
|
|
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,
|
|
standard: dict[str, Any] | None = None,
|
|
source: str | None = None,
|
|
) -> tuple[str, str]:
|
|
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.
|
|
|
|
`source` = 그 엔진 소스 지문의 키(없으면 지금 코드 · `read_stored_any`)."""
|
|
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),
|
|
json.dumps(standard, sort_keys=True, default=str) if standard else None,
|
|
ALGO_VERSION,
|
|
source or source_fingerprint(),
|
|
)
|
|
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,
|
|
saved: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
"""지킬 결과 — 기본값(빈 제약 · 50/25/25 · 설계 지형) 또는 저장된 제약(`initial_choice`) 그대로.
|
|
|
|
모달이 여는 값이라 「최근 2벌」 정리에서 따로 센다(351-1 · 사용자 시도에 안 밀림).
|
|
"""
|
|
if terrain is not None:
|
|
return False
|
|
filled = {key: value for key, value in (criteria or {}).items() if value is not None}
|
|
used = resolve_weights(mode, weights)
|
|
if not filled and (mode != "weighted" or used == DEFAULT_WEIGHTS):
|
|
return True
|
|
saved = saved or {}
|
|
return filled == (saved.get("criteria") or {}) and (
|
|
mode != "weighted" or used == saved.get("weights")
|
|
)
|
|
|
|
|
|
def _current(data: Any) -> bool:
|
|
return (
|
|
isinstance(data, dict)
|
|
and data.get("version") == ALGO_VERSION
|
|
and data.get("source") == source_fingerprint()
|
|
)
|
|
|
|
|
|
def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
|
|
"""저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None · 파일을 안 건드림(505-1)."""
|
|
path = store_dir(project_root) / f"{key}.json"
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
return data if _current(data) else None
|
|
|
|
|
|
def read_stored_any(
|
|
project_root: Path, pool: str, key_of: Callable[[str], str]
|
|
) -> tuple[str, dict[str, Any]] | None:
|
|
"""(키, 저장본) — 같은 판 · 묶음이고 그 파일의 소스 지문으로 다시 낸 키(`key_of`)가 파일 이름과
|
|
같은 것 중 최근 것 · 없으면 None. 프로젝트 접근 전용(505-1 덧 · 엔진 고쳐도 안내 안 쏟아짐)."""
|
|
found: list[tuple[float, str, dict[str, Any]]] = []
|
|
for path in store_dir(project_root).glob("*.json"):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
mtime = path.stat().st_mtime
|
|
except (OSError, ValueError):
|
|
continue
|
|
if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
|
|
continue
|
|
if data.get("pool") == pool and key_of(str(data.get("source"))) == path.stem:
|
|
found.append((mtime, path.stem, data))
|
|
if not found:
|
|
return None
|
|
_, key, data = max(found, key=lambda item: item[0])
|
|
return key, 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,
|
|
"source": source_fingerprint(),
|
|
"key": key,
|
|
"pool": pool,
|
|
"default": default,
|
|
"created_at": datetime.now().isoformat(timespec="seconds"),
|
|
"request": request,
|
|
"result": result,
|
|
},
|
|
)
|
|
prune_stored(project_root, request, default)
|
|
return True
|
|
|
|
|
|
def _kind(request: Any, default: Any) -> tuple[Any, Any, bool]:
|
|
request = request if isinstance(request, dict) else {}
|
|
return request.get("mode"), request.get("terrain"), bool(default)
|
|
|
|
|
|
def prune_stored(project_root: Path, request: dict[str, Any], default: bool = False) -> int:
|
|
"""같은 종류(갈래 · 지형 · 기본값 여부) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다.
|
|
|
|
기본값 · 저장된 제약 결과(`default`)는 따로 셈 — 사용자 시도가 밀어내지 않음(351-1). 같은 판 ·
|
|
소스 지문 안에서만 셈 — 다른 지문(다른 PC · 워크트리 코드) 파일은 안 건드림(420-2). 지운 수.
|
|
"""
|
|
kind = _kind(request, default)
|
|
same: list[tuple[int, Path]] = []
|
|
for path in store_dir(project_root).glob("*.json"):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if _current(data) and _kind(data.get("request"), data.get("default")) == kind:
|
|
same.append((path.stat().st_mtime_ns, path))
|
|
except (OSError, ValueError, AttributeError):
|
|
continue
|
|
same.sort(reverse=True)
|
|
for _, path in same[KEEP_PER_KIND:]:
|
|
path.unlink(missing_ok=True)
|
|
return max(0, len(same) - KEEP_PER_KIND)
|
|
|
|
|
|
def _remove_where(project_root: Path, drop: Any) -> int:
|
|
"""저장본 중 `drop(data, path)` 가 참인 파일을 지운다(못 읽는 파일 = data None) — 지운 수."""
|
|
removed = 0
|
|
for path in store_dir(project_root).glob("*.json"):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
data = None
|
|
if drop(data, path):
|
|
path.unlink(missing_ok=True)
|
|
removed += 1
|
|
return removed
|
|
|
|
|
|
def purge_outdated(project_root: Path, now: float | None = None) -> int:
|
|
"""다른 판 · 소스 지문 저장본 중 `STALE_DAYS` 지난 것만 지운다(미리 셈 첫머리 · 420-2).
|
|
|
|
storage 를 두 PC · 여러 워크트리가 같이 써 코드가 다르면 지문이 다름 — 새것은 안 읽고 둠.
|
|
지운 수."""
|
|
limit = (time.time() if now is None else now) - STALE_DAYS * 86400.0
|
|
|
|
def stale(data: Any, path: Path) -> bool:
|
|
try:
|
|
return not _current(data) and path.stat().st_mtime < limit
|
|
except OSError:
|
|
return False
|
|
|
|
return _remove_where(project_root, stale)
|
|
|
|
|
|
def drop_pool_results(project_root: Path, pool: str) -> int:
|
|
"""[재계산] — 이 후보 묶음(다섯 갈래 · 종단 켬끔)의 저장본을 지우고 메모리 묶음도 버린다.
|
|
|
|
`pool_lock(pool)` 안에서 부름(미리 셈이 다시 써 버리지 않게) · 지금 판 · 소스 지문 것만
|
|
(다른 지문 = 다른 PC · 워크트리 것 · 420-2). 지운 수."""
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
|
|
|
|
_POOLS.pop(bytes.fromhex(pool), None)
|
|
return _remove_where(
|
|
project_root, lambda data, _path: _current(data) and data.get("pool") == pool
|
|
)
|
|
|
|
|
|
def forget_pool_on_standard(pool: str, standard: dict[str, Any] | None) -> None:
|
|
"""메모리 묶음 후보에 붙은 균형 종단은 그때 표준 횡단면으로 셈 — 바뀌면 묶음을 버린다."""
|
|
from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
|
|
|
|
mark = json.dumps(standard, sort_keys=True, default=str)
|
|
if _POOL_STANDARDS.get(pool, mark) != mark:
|
|
_POOLS.pop(bytes.fromhex(pool), None)
|
|
_POOL_STANDARDS[pool] = mark
|
|
|
|
|
|
@asynccontextmanager
|
|
async def pool_lock(pool: str) -> AsyncIterator[None]:
|
|
# ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로
|
|
async with _LOCKS.setdefault(pool, asyncio.Lock()):
|
|
yield
|