- `file_digest()`(derived_store) — blake2b 내용 지문 · 한 프로세스 안 경로 · 시각 · 크기로 기억
- 비용면 서명 · 계류 지문 · 초기 종단 `stat_marks` = 내용 지문(옮겨 심기 · 동기화 · 다시 짓기로 시각만 바뀌면 같은 열쇠)
- 초기 종단 표식 = 확정 지형이 읽는 파일만(`_source_npz_paths` · `*_{필터}*.npz` glob 걷음)
- `balanced_profile_of` 에 selection — 추천 초기 종단 목표 계획고 = 지반선과 같은 고른 지형(3h)
- 열쇠 꼴이 바뀌어 기존 추천 · 초기 종단 저장본은 다시 셈(763-5 몫)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""763-2 같은 서피스 = 내용 지문 — 시각만 바뀌면 같은 열쇠 · 한 바이트 바꿈 = 다른 열쇠 ·
|
|
초기 종단 표식 = 확정 지형이 읽는 파일만 · 초기 종단 목표 = 고른 지형.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_763_2_content_key.py -q
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from test_763_1_confirmed_terrain import _models
|
|
|
|
from B05_Profile import B05_Profile_Engine_InitialProfile as ip
|
|
from B05_Profile import B05_Profile_Engine_Solver as solver
|
|
from B05_Profile import B05_Profile_Router_Replan as replan
|
|
from B05_Profile import B05_Profile_Router_RouteInitial as router
|
|
from common_util.common_util_derived_store import file_digest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _retime(folder: Path, seconds: int) -> None:
|
|
for path in folder.iterdir():
|
|
os.utime(path, (seconds, seconds))
|
|
|
|
|
|
def _flip_byte(path: Path) -> None:
|
|
data = bytearray(path.read_bytes())
|
|
data[-1] ^= 0xFF
|
|
path.write_bytes(bytes(data))
|
|
|
|
|
|
def test_파일_지문_내용만(tmp_path) -> None:
|
|
a, b = tmp_path / "a.bin", tmp_path / "b.bin"
|
|
a.write_bytes(b"terrain" * 1000)
|
|
b.write_bytes(b"terrain" * 1000)
|
|
os.utime(b, (1_000_000, 1_000_000))
|
|
assert file_digest(a) == file_digest(b)
|
|
_flip_byte(b)
|
|
assert file_digest(a) != file_digest(b)
|
|
assert file_digest(tmp_path / "none.bin") is None
|
|
|
|
|
|
def test_비용면_서명_시각만_바뀌면_같음(tmp_path) -> None:
|
|
(tmp_path / "one").mkdir()
|
|
first = _models(tmp_path / "one")
|
|
for method in ("implicit", "dtm"):
|
|
solver._source_npz_paths(first, "classification", method, False) # 굳힌 격자 지음
|
|
second = tmp_path / "two"
|
|
shutil.copytree(first, second)
|
|
_retime(second, 2_000_000_000) # 옮겨 심기 · 동기화 = 시각만 바뀜
|
|
for method in ("implicit", "dtm"):
|
|
one = solver._cost_surface_signature(first, "classification", method, False)
|
|
two = solver._cost_surface_signature(second, "classification", method, False)
|
|
assert one == two
|
|
_flip_byte(second / "dtm_classification.npz")
|
|
assert solver._cost_surface_signature(
|
|
first, "classification", "dtm", False
|
|
) != solver._cost_surface_signature(second, "classification", "dtm", False)
|
|
# 초기 종단 표식도 내용
|
|
names = ["implicit_classification.npz", "implicit_classification_grid.npz"]
|
|
assert ip.stat_marks([first / n for n in names]) == ip.stat_marks([second / n for n in names])
|
|
|
|
|
|
def test_초기_종단_표식_확정_지형_파일만() -> None:
|
|
source = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial_Initial.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert "models.glob(" not in source
|
|
assert '"models": ip.stat_marks(\n _source_npz_paths(' in source
|
|
# 목표 = 고른 지형(지반선 표본기와 같은 selection)
|
|
assert "request.criteria, curves, selection\n" in source
|
|
|
|
|
|
def test_균형_종단은_받은_지형으로(monkeypatch) -> None:
|
|
import B05_Profile.B05_Profile_Engine_Grade as grade
|
|
import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
|
|
import common_util.common_util_surface_sampler as sampler
|
|
|
|
opened: list[tuple] = []
|
|
|
|
async def road(_pid):
|
|
return "간선", 20, "일반"
|
|
|
|
async def standard(_pid):
|
|
return None
|
|
|
|
def build(_models, filter_key, method, smooth):
|
|
opened.append((filter_key, method, smooth))
|
|
return object()
|
|
|
|
def route_profile(*_a, **_k):
|
|
return {
|
|
"stations_m": [0.0],
|
|
"plan_m": [0.0],
|
|
"cut_m3": 0,
|
|
"fill_m3": 0,
|
|
"mass_end_m3": 0,
|
|
"seconds": 0,
|
|
}
|
|
|
|
def no_db():
|
|
raise AssertionError("고른 지형이 있으면 확정 params 를 다시 읽지 않음")
|
|
|
|
monkeypatch.setattr(replan, "_road_settings", road)
|
|
monkeypatch.setattr(router, "project_standard", standard)
|
|
monkeypatch.setattr(router, "get_db_pool", no_db)
|
|
monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
|
|
monkeypatch.setattr(sampler, "build_surface_sampler", build)
|
|
monkeypatch.setattr(profile, "route_profile", route_profile)
|
|
sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
|
|
points = [{"x": 0.0, "y": 0.0}, {"x": 10.0, "y": 0.0}]
|
|
got = asyncio.run(router.balanced_profile_of(uuid4(), Path("."), points, None, None, sheet))
|
|
assert got is not None and opened == [("sheet_laplace", "dtm", True)]
|