Files
Aislo/B03_FileInput/B03_FileInput_Service_WF1.py
T
eomsangdonandClaude Opus 5 72075563e3 feat(B03·B04): 지형 라이다 여러 장 입력·병합 전처리
- 지형 파일 개수 제한 해제(정확히 1개 → 1장 이상), 한 카드에 여러 장 담기
  (화면 표시 「용화_서편.las 외 1장」, 업로드는 한 장씩 차례로 전송)
- 구조화 엔진이 여러 파일을 합친 범위로 한 벌 생성 — WF1 자동 전처리·B04 재분석 모두
  프로젝트 지형 파일 전부를 대상으로 실행
- 점이 1억 개를 넘으면 씨닝 — 지면 분류점은 전부 남기고 나머지만 0.5m 칸 최저점으로 축소
  (용화 실측: 4,900만점 → 249만점, 지면점 1,140,716개 그대로, 1m 지면격자 표고차 0.0000m)
- 머리글만 읽어 5km 넘게 떨어진 파일은 업로드 거부 (임도는 길어도 2~3km)
- 화면 조립부 700줄 준수를 위해 terrainCoverage 를 판정 모듈로 이동

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-06 17:45:59 +09:00

253 lines
12 KiB
Python

"""B03 업로드 이후 WF1 백그라운드 분석·자동 확정 서비스."""
import asyncio
import logging
from pathlib import Path
from typing import Any
from uuid import UUID
import aiomysql
from B03_FileInput.B03_FileInput_Email import (
send_analysis_error_email,
send_initial_analysis_complete_email,
send_initial_design_failed_email,
)
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from common_util.common_util_initial_snapshot import is_design_failed, read_design_failure
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_surface_confirmation import surface_confirmation_defaults
from common_util.common_util_workflow_state import fail_stage, start_stage
from config.config_db import get_db_pool
from config.config_system import (
AUTO_DESIGN_CHAIN_ENABLED,
SEND_ANALYSIS_COMPLETION_EMAIL,
SHEET_SURFACE_DEFAULT_METHOD,
SURFACE_AUTO_METHODS,
SURFACE_CONTOUR_INTERVAL_M,
)
logger = logging.getLogger(__name__)
async def _get_project_notification_info(
connection: aiomysql.Connection,
project_id: UUID,
) -> dict[str, Any] | None:
async with connection.cursor(aiomysql.DictCursor) as cursor:
await cursor.execute(
"""
SELECT p.id, p.name AS project_name, u.email AS user_email, u.name AS user_name
FROM projects p
JOIN users u ON u.id = p.user_id
WHERE p.id = %s AND p.deleted_at IS NULL AND u.deleted_at IS NULL
""",
(str(project_id),),
)
row = await cursor.fetchone()
return dict(row) if row else None
async def trigger_wf1_analysis_and_email(
*,
project_id: UUID,
input_file_id: int,
user_role: str,
) -> None:
"""WF1 분석·DB 저장 후 자동 확정하고 이메일을 발송한다.
자동 확정은 역할 무관이다(2026-08-04 사용자 확정 — 관리자도 같은 사용자).
`user_role`은 호출부 시그니처 유지용으로만 남겨 둔다.
"""
_ = user_role
pool = get_db_pool()
project_info: dict[str, Any] | None = None
try:
# 자동 전처리는 기본 조합만 만든다 — 필터는 엔진이 입력 LAS를 보고 정하고
# (분류 있으면 classification, 없으면 csf), 표현은 SURFACE_AUTO_METHODS.
# 다른 조합은 관리자가 B04 드롭다운에서 요청할 때 만든다 (2026-09-01 사용자 확정).
methods = list(SURFACE_AUTO_METHODS)
params = {
"input_file_id": str(input_file_id),
"source_filters": None,
"methods": methods,
"force": False,
}
async with pool.acquire() as connection:
async with connection.cursor() as cursor:
await start_stage(cursor, str(project_id), 1, params)
await connection.commit()
stored_path = await get_project_storage_relative_path(connection, project_id)
project_info = await _get_project_notification_info(connection, project_id)
from B04_PreProcess.B04_PreProcess_Repository import (
get_input_file,
list_project_point_cloud_paths,
)
input_file = await get_input_file(connection, project_id, input_file_id)
project_root = Path(resolve_stored_project_path(stored_path))
# 지형 파일이 여러 장이면 합쳐서 한 벌로 전처리한다(2026-09-06 사용자 확정).
terrain_paths = await list_project_point_cloud_paths(
connection, project_id, project_root
)
source_path = project_root / Path(str(input_file["raw_file_path"]))
# LAS 없는 설계(2026-08-30): 입력이 계획노선 CSV면 도엽등고선 서피스 분석으로 간다.
las_free = str(input_file.get("file_type") or "").lower() not in {"las", "laz"}
if not source_path.is_file():
raise FileNotFoundError(
"계획 노선 파일을 찾을 수 없습니다."
if las_free
else "원본 LAS/LAZ 파일을 찾을 수 없습니다."
)
from B04_PreProcess.B04_PreProcess_Engine import run_surface_analysis
from B04_PreProcess.B04_PreProcess_Repository import save_surface_analysis_to_db
from B04_PreProcess.B04_PreProcess_Router import write_surface_progress
write_surface_progress(project_root, 5, "analyzing", "WF1 분석을 시작합니다.")
def _on_progress(percent: int, stage: str, message: str) -> None:
write_surface_progress(project_root, percent, stage, message)
if las_free:
from B04_PreProcess.B04_PreProcess_Engine_SheetSurface import (
run_sheet_surface_analysis,
)
analysis_result = await asyncio.to_thread(
run_sheet_surface_analysis,
project_root,
source_path,
on_progress=_on_progress,
)
else:
analysis_result = await asyncio.to_thread(
run_surface_analysis,
project_root,
terrain_paths or [source_path],
source_filters=None,
methods=methods,
force=False,
on_progress=_on_progress,
)
# 엔진이 정한 기본 필터 — LAS 없는 설계(도엽 서피스)는 이 키가 없다.
resolved_filters = list(analysis_result.get("source_filters") or [])
if not resolved_filters and las_free:
resolved_filters = [f"sheet_{SHEET_SURFACE_DEFAULT_METHOD}"]
auto_confirmation_error: str | None = None
auto_confirmed = False
confirmed_model_id: int | None = None
async with pool.acquire() as connection:
await connection.begin()
try:
await save_surface_analysis_to_db(
connection,
project_id=project_id,
input_file_id=input_file_id,
analysis_result=analysis_result,
source_filters=resolved_filters,
)
# 역할 무관 자동 확정(2026-08-04 사용자 확정) — 시스템 관리자도 같은
# 사용자다. 모두 기본값으로 자동 확정하고 B05·B06 자동 계산 체인까지 탄다.
# 관리자는 B04에서 다른 모델을 골라 재확정할 수 있다(체인은 경로가 이미
# 있으면 스스로 건너뛴다).
from B04_PreProcess.B04_PreProcess_Service import (
confirm_surface_selection,
find_surface_model_for_selection,
)
# LAS 없는 설계는 도엽 서피스 모델(sheet/dtm)로 확정한다.
# 스무딩은 LAS 경로와 같이 적용한다(2026-08-30 사용자 확정) — 방식마다
# `dtm_sheet_*_smooth.npz`를 같이 만들어 두므로 종·횡단이 그걸 샘플링한다.
if las_free:
selection = {
"source_filter": f"sheet_{SHEET_SURFACE_DEFAULT_METHOD}",
"method": "dtm",
"smooth": True,
"contour_interval_m": SURFACE_CONTOUR_INTERVAL_M,
}
else:
# 자동 확정은 엔진이 정한 기본 필터를 그대로 따른다 — 이제 그 필터
# 하나만 만들어 두므로 config 기본값과 어긋나면 확정할 모델이 없다.
selection = surface_confirmation_defaults()
if resolved_filters:
selection["source_filter"] = resolved_filters[0]
selection["method"] = methods[0]
try:
model_id = await find_surface_model_for_selection(
connection, project_id, selection
)
except LookupError as exc:
auto_confirmation_error = str(exc)
logger.error(
"WF1 자동 확정 보류: project_id=%s reason=%s",
project_id,
auto_confirmation_error,
)
else:
await confirm_surface_selection(connection, project_id, model_id, selection)
auto_confirmed = True
confirmed_model_id = model_id
await connection.commit()
except Exception:
await connection.rollback()
raise
if auto_confirmation_error:
progress_stage = "awaiting_confirmation"
progress_message = f"WF1 자동 확정 보류 — {auto_confirmation_error}"
elif auto_confirmed:
progress_stage = "completed"
progress_message = "지표면 모델 자동 확정 완료 — 노선 설계로 이동하세요."
else:
progress_stage = "awaiting_confirmation"
progress_message = "WF1 분석이 완료되었습니다. 사용할 모델을 확정하세요."
write_surface_progress(project_root, 100, progress_stage, progress_message)
# WF1 자동 확정까지 끝났으면 같은 백그라운드 태스크에서 B05 기본 경로 → B06 기본
# 횡단 설계 체인을 잇는다(2026-08-04 사용자 확정). 체인은 실패를 스스로 격리한다.
design_summary: dict[str, Any] | None = None
if auto_confirmed and AUTO_DESIGN_CHAIN_ENABLED:
from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
design_summary = await run_auto_design_chain(
project_id, surface_model_id=confirmed_model_id
)
# 알림 메일은 여기 한 통뿐이다 — 업로드 직후와 분석 직후로 나눠 두 통을 보내던 것을
# 초기 설계(B04~B06)까지 마친 시점의 통합 메일로 합쳤다(2026-08-08 사용자 지시).
# 체인이 깨졌으면 완료 메일 대신 관리자 문의 안내를 보낸다 — 부분 결과는 분석
# 안 됨과 다르지 않다(2026-09-02 사용자 확정).
design_failure = read_design_failure(project_root) if is_design_failed(project_root) else ""
if SEND_ANALYSIS_COMPLETION_EMAIL and project_info and project_info.get("user_email"):
if design_failure:
await send_initial_design_failed_email(
project_id=project_id,
project_name=str(project_info.get("project_name") or project_id),
to_email=str(project_info["user_email"]),
reason=design_failure,
)
else:
await send_initial_analysis_complete_email(
project_id=project_id,
project_name=str(project_info.get("project_name") or project_id),
to_email=str(project_info["user_email"]),
analysis_result=analysis_result,
design_summary=design_summary,
)
except Exception as exc:
logger.exception("WF1 백그라운드 분석 실패: project_id=%s", project_id)
async with pool.acquire() as connection, connection.cursor() as cursor:
await fail_stage(cursor, str(project_id), 1, str(exc))
await connection.commit()
if project_info and project_info.get("user_email"):
await send_analysis_error_email(
project_id=project_id,
project_name=str(project_info.get("project_name") or project_id),
to_email=str(project_info["user_email"]),
error_message=str(exc),
)