eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix(532-3): E02 측점(B06 암 경계)을 E01 동결 복사본 db.json 에서 읽음 — 원본 임도 DB 안 읽음 · 복사본 없으면 빈 목록

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
eomsangdon
2026-10-25 20:07:38 +09:00
co-authored by Claude Opus 5.5
parent 333d4f63ee
commit deaacc1100
2 changed files with 64 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
E02_StabGround/E02_StabGround_Router.py
+36 -48
View File
@@ -5,7 +5,7 @@
· 측점(B06 암 경계)은 서버 값
- `POST …/stab/ground/reset` — [초기화] 초기값을 작업본에 덮어씀(재계산 아님)
초기값 = E 생성 직후 마스터 기본값 + B06 측점(`ensure_initial`) · 그 밖에는 쓰지 않음.
초기값 = E 생성 직후 마스터 기본값 + 복사본 B06 측점(`ensure_initial`) · 그 밖에는 쓰지 않음.
공유 회사(보기만)는 `require_project_access` 가 PUT · POST 를 막음.
"""
@@ -21,12 +21,13 @@ from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Repository import get_latest_route
from common_util.common_util_auth import verify_session
from common_util.common_util_json import atomic_write_json
from common_util.common_util_project_copy import MANIFEST_NAME
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
from config.config_system_design import STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M
from E01_StabRegister.E01_StabRegister_Source import db_snapshot, source_dir
from E02_StabGround import E02_StabGround_Model as model
router = APIRouter(prefix="/api/projects", tags=["E02 Stab Ground"])
@@ -36,38 +37,30 @@ class GroundRequest(BaseModel):
ground: dict[str, Any]
async def _b06_stations(connection: aiomysql.Connection, project_ids: list[str]) -> list[dict]:
def _b06_stations(project_root: Path) -> list[dict]:
"""측점 · 지반 라벨 · 토층 두께(암 측점 = 암 경계 깊이 · 토사 = 없음).
자기 복사본(532-1)이 먼저 · 없으면 원본 임도."""
for project_id in project_ids:
route = await get_latest_route(connection, UUID(project_id))
if not route:
continue
async with connection.cursor() as cursor:
await cursor.execute(
"SELECT chainage_m, data FROM cross_sections WHERE route_id = %s"
" ORDER BY chainage_m",
(route["id"],),
)
rows = await cursor.fetchall()
stations = []
for chainage, data in rows:
data = json.loads(data) if isinstance(data, str) else data or {}
design = data.get("design") or {}
ground_type = design.get("ground_type") or "soil"
offset = design.get("rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M)
rock = ground_type in model.ROCK_LABELS and offset is not None
stations.append(
{
"chainage_m": round(float(chainage), 3),
"ground_type": ground_type,
"soil_thickness_m": round(abs(float(offset)), 3) if rock else None,
}
)
if stations:
return stations
return []
E01 동결 복사본(`E01_StabRegister/source/db.json`)만 읽음 — 복사본이 없으면 빈 목록.
"""
root = source_dir(project_root)
if not (root / MANIFEST_NAME).is_file():
return []
rows = sorted(db_snapshot(root)["cross_sections"], key=lambda r: float(r["chainage_m"] or 0))
stations = []
for row in rows:
data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"] or {}
design = data.get("design") or {}
ground_type = design.get("ground_type") or "soil"
offset = design.get("rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M)
rock = ground_type in model.ROCK_LABELS and offset is not None
stations.append(
{
"chainage_m": round(float(row["chainage_m"]), 3),
"ground_type": ground_type,
"soil_thickness_m": round(abs(float(offset)), 3) if rock else None,
}
)
return stations
async def _project(project_id: UUID) -> tuple[Path, dict[str, Any]]:
@@ -78,7 +71,7 @@ async def _project(project_id: UUID) -> tuple[Path, dict[str, Any]]:
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.") from error
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"SELECT company_id, project_kind, source_project_id FROM projects WHERE id = %s",
"SELECT company_id, project_kind FROM projects WHERE id = %s",
(str(project_id),),
)
project = await cursor.fetchone()
@@ -87,23 +80,20 @@ async def _project(project_id: UUID) -> tuple[Path, dict[str, Any]]:
return Path(resolve_stored_project_path(stored)), project
async def ensure_initial(project_root: Path, project_id: str, source_id: str | None) -> dict:
"""초기값이 없을 때만 한 번 — 기본값 + B06 측점(E 생성 직후 · 옛 E 프로젝트는 첫 GET)."""
def ensure_initial(project_root: Path) -> dict:
"""초기값이 없을 때만 한 번 — 기본값 + 복사본 B06 측점(E 생성 직후 · 옛 E 프로젝트는 첫 GET)."""
path = model.initial_path(project_root)
if path.is_file():
return json.loads(path.read_text(encoding="utf-8"))
async with get_db_pool().acquire() as connection:
ids = [project_id] + ([source_id] if source_id else [])
stations = await _b06_stations(connection, ids)
ground = model.build_initial(model.load_defaults(), stations)
ground = model.build_initial(model.load_defaults(), _b06_stations(project_root))
atomic_write_json(path, ground)
return ground
async def seed_initial(project_id: str) -> None:
"""E01 생성 직후 부름 — 초기값 한 벌."""
root, project = await _project(UUID(project_id))
await ensure_initial(root, project_id, project["source_project_id"])
root, _ = await _project(UUID(project_id))
ensure_initial(root)
def _read_working(project_root: Path) -> dict[str, Any] | None:
@@ -120,7 +110,7 @@ async def get_ground(
project_id: UUID, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
root, project = await _project(project_id)
initial = await ensure_initial(root, str(project_id), project["source_project_id"])
initial = ensure_initial(root)
working = _read_working(root)
return {
"ground": working or initial,
@@ -132,16 +122,14 @@ async def get_ground(
@router.put("/{project_id}/stab/ground")
async def put_ground(project_id: UUID, payload: GroundRequest) -> dict[str, Any]:
root, project = await _project(project_id)
root, _ = await _project(project_id)
defaults = model.load_defaults()
errors = model.check_ground(payload.ground, defaults["limits"])
if errors:
raise HTTPException(status_code=422, detail={"errors": errors})
ground = model.clean(payload.ground)
# 측점(B06 암 경계)은 서버 값 — 브라우저 값을 받아 적지 않음
base = _read_working(root) or await ensure_initial(
root, str(project_id), project["source_project_id"]
)
base = _read_working(root) or ensure_initial(root)
ground["stations"] = base.get("stations", [])
ground["version"] = defaults["version"]
atomic_write_json(model.ground_path(root), ground)
@@ -150,7 +138,7 @@ async def put_ground(project_id: UUID, payload: GroundRequest) -> dict[str, Any]
@router.post("/{project_id}/stab/ground/reset")
async def reset_ground(project_id: UUID) -> dict[str, Any]:
root, project = await _project(project_id)
initial = await ensure_initial(root, str(project_id), project["source_project_id"])
root, _ = await _project(project_id)
initial = ensure_initial(root)
atomic_write_json(model.ground_path(root), initial)
return {"ground": initial}
resources/tester/test_532_3_e02_ground.py
+28 -17
View File
@@ -183,28 +183,33 @@ def test_측점_적용():
assert model.resolve_stations(RESOLVE_CASES[2])[0]["lower"] == "soft_rock"
def _write_copy(root: Path) -> None:
"""E01 동결 복사본 꼴(copy_manifest.json + db.json cross_sections) — 측점은 여기서만 읽음."""
source = root / "E01_StabRegister" / "source"
source.mkdir(parents=True)
(source / "copy_manifest.json").write_text("{}", encoding="utf-8")
rows = [
{"chainage_m": 20.0, "data": json.dumps({"design": {"ground_type": "ripping_rock"}})},
{"chainage_m": 0.0, "data": {"design": {"ground_type": "soil"}}},
{
"chainage_m": 60.0,
"data": {"design": {"ground_type": "ripping_rock", "rock_boundary_offset_m": -0.5}},
},
{
"chainage_m": 40.0,
"data": {"design": {"ground_type": "blasting_rock", "rock_boundary_offset_m": -0.8}},
},
]
(source / "db.json").write_text(json.dumps({"cross_sections": rows}), encoding="utf-8")
@pytest.fixture
def project(tmp_path, monkeypatch):
async def fake_project(_project_id):
return tmp_path, {"company_id": 1, "project_kind": "STABILITY", "source_project_id": None}
async def fake_stations(_connection, _ids):
return copy.deepcopy(STATIONS)
class FakePool:
def acquire(self):
class Ctx:
async def __aenter__(self):
return None
async def __aexit__(self, *_):
return False
return Ctx()
return tmp_path, {"company_id": 1, "project_kind": "STABILITY"}
monkeypatch.setattr(router, "_project", fake_project)
monkeypatch.setattr(router, "_b06_stations", fake_stations)
monkeypatch.setattr(router, "get_db_pool", lambda: FakePool())
_write_copy(tmp_path)
return tmp_path
@@ -276,6 +281,12 @@ def test_저장_다시읽음_초기화(project):
assert json.loads(model.ground_path(project).read_text(encoding="utf-8")) == initial
def test_복사본_없으면_측점_없음(tmp_path):
assert router._b06_stations(tmp_path) == []
_write_copy(tmp_path)
assert router._b06_stations(tmp_path) == STATIONS # 누가 차례 · 암 경계 없으면 −0.5
def test_다른_회사는_보기만(project):
from uuid import UUID