Import S-CANVAS source + iter=1~7 lint cleanup

S-CANVAS (Saman Corp.) — DXF + DEM + AI 기반 3D 조감도 생성 엔진.
~24k LOC Python (scanvas_maker.py 7072 LOC GUI + 구조물 파서/빌더 다수).

이 커밋은 7-iter cleanup이 적용된 상태로 import:
- F821 8 + B023 6: 비동기 lambda + except/loop 변수 캡처 NameError
  (Py3.13에서 reproduce 확인된 진짜 버그)
- RUF012 4 + RUF013 1: ClassVar / implicit Optional 명시화
- F811/B905/B904/F401/F841/W293/F541/UP/SIM/RUF/PLR 700+ cleanup/modernization

신규 파일:
- ruff.toml: target=py313, Korean unicode/저자 스타일/도메인 복잡도 무력화
- requirements-py313.txt: pyproj>=3.7, scipy>=1.14, numpy>=2.0.2 (Py3.13 wheel)
- .gitignore: gcp-key.json, 캐시, 백업, 생성 이미지 제외

검증: ruff 0 errors, py_compile 0 errors, import 33/33 OK on Py3.13.13.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-08 10:29:08 +09:00
parent 53d8b53c2f
commit b9342f6726
92 changed files with 3413501 additions and 0 deletions

View File

View File

@@ -0,0 +1,137 @@
"""로거 - SQLite DB + structlog 기반 작업 이력 추적."""
from __future__ import annotations
import logging
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
import structlog
from sqlalchemy import Column, DateTime, Float, Integer, String, Text, create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
# ──────────────────────────── ORM 모델 ────────────────────────────
class Base(DeclarativeBase):
pass
class JobRecord(Base):
"""조감도 생성 작업 1건의 이력 레코드."""
__tablename__ = "jobs"
id = Column(Integer, primary_key=True, autoincrement=True)
dxf_path = Column(String(512), nullable=False)
dxf_hash = Column(String(32))
timestamp = Column(DateTime, default=datetime.utcnow)
seed = Column(Integer)
prompt_version = Column(String(32))
prompt_hash = Column(String(32))
status = Column(String(16), default="pending") # pending / running / done / failed
output_path = Column(String(512))
quality_score = Column(Float)
error_message = Column(Text)
latency_ms = Column(Float)
# ──────────────────────────── DB 세션 ────────────────────────────
_engine = None
_SessionFactory = None
def init_db(db_path: str | Path = "cad_aerial_gen.db"):
global _engine, _SessionFactory
_engine = create_engine(f"sqlite:///{db_path}", echo=False)
Base.metadata.create_all(_engine)
_SessionFactory = sessionmaker(bind=_engine)
def get_db_session() -> Session:
if _SessionFactory is None:
init_db()
return _SessionFactory()
# ──────────────────────────── structlog 설정 ────────────────────────────
def setup_logging(log_file: Optional[Path] = None, level: str = "INFO"):
"""콘솔 + 파일 동시 로깅을 설정한다."""
handlers = [logging.StreamHandler(sys.stdout)]
if log_file:
log_file.parent.mkdir(parents=True, exist_ok=True)
handlers.append(logging.FileHandler(str(log_file), encoding="utf-8"))
logging.basicConfig(
format="%(message)s",
level=getattr(logging, level.upper(), logging.INFO),
handlers=handlers,
)
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S"),
structlog.dev.ConsoleRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(
getattr(logging, level.upper(), logging.INFO)
),
logger_factory=structlog.PrintLoggerFactory(),
)
def get_logger(name: str = "cad_aerial_gen"):
return structlog.get_logger(name)
# ──────────────────────────── 작업 이력 헬퍼 ────────────────────────────
class JobLogger:
"""작업 이력 CRUD 래퍼."""
def create_job(self, db: Session, dxf_path: str, dxf_hash: str = "") -> JobRecord:
record = JobRecord(dxf_path=dxf_path, dxf_hash=dxf_hash, status="pending")
db.add(record)
db.commit()
db.refresh(record)
return record
def start_job(self, db: Session, job_id: int, seed: int, prompt_version: str, prompt_hash: str):
record = db.query(JobRecord).filter_by(id=job_id).first()
if record:
record.status = "running"
record.seed = seed
record.prompt_version = prompt_version
record.prompt_hash = prompt_hash
db.commit()
def complete_job(
self,
db: Session,
job_id: int,
output_path: str,
quality_score: float,
latency_ms: float,
):
record = db.query(JobRecord).filter_by(id=job_id).first()
if record:
record.status = "done"
record.output_path = output_path
record.quality_score = quality_score
record.latency_ms = latency_ms
db.commit()
def fail_job(self, db: Session, job_id: int, error: str):
record = db.query(JobRecord).filter_by(id=job_id).first()
if record:
record.status = "failed"
record.error_message = error
db.commit()
def list_jobs(self, db: Session, limit: int = 50):
return db.query(JobRecord).order_by(JobRecord.id.desc()).limit(limit).all()

View File

@@ -0,0 +1,58 @@
"""프롬프트 레지스트리 - 버전 관리 및 재현 가능성 보장."""
from __future__ import annotations
from pathlib import Path
from typing import Dict, List, Optional
import yaml
class PromptRegistry:
"""프롬프트 템플릿 버전을 관리하고 변경 이력을 추적한다."""
def __init__(self, templates_dir: Path):
self.templates_dir = templates_dir
def list_versions(self) -> List[str]:
"""사용 가능한 템플릿 버전 목록을 반환한다 (최신순)."""
yamls = sorted(self.templates_dir.glob("v*.yaml"), reverse=True)
return [p.stem for p in yamls]
def latest_version(self) -> Optional[str]:
versions = self.list_versions()
return versions[0] if versions else None
def load_template(self, version: str) -> Dict:
path = self.templates_dir / f"{version}.yaml"
if not path.exists():
raise FileNotFoundError(f"템플릿 버전 {version}이 없습니다.")
with open(path, encoding="utf-8") as f:
return yaml.safe_load(f)
def compare(self, version_a: str, version_b: str) -> Dict:
"""두 버전의 차이점을 반환한다."""
a = self.load_template(version_a)
b = self.load_template(version_b)
diff = {}
all_keys = set(a) | set(b)
for key in all_keys:
va, vb = a.get(key), b.get(key)
if va != vb:
diff[key] = {"old": va, "new": vb}
return diff
def save_new_version(self, new_version: str, template: Dict) -> Path:
"""새 버전 템플릿을 저장한다."""
path = self.templates_dir / f"{new_version}.yaml"
if path.exists():
raise FileExistsError(f"버전 {new_version}이 이미 존재합니다.")
with open(path, "w", encoding="utf-8") as f:
yaml.dump(template, f, allow_unicode=True, default_flow_style=False)
return path
def get_version_for_hash(self, prompt_hash: str, db_session) -> Optional[str]:
"""프롬프트 해시로 사용된 버전을 역조회한다."""
from harness.logger import JobRecord
record = db_session.query(JobRecord).filter_by(prompt_hash=prompt_hash).first()
return record.prompt_version if record else None

View File

@@ -0,0 +1,98 @@
"""품질 검증기 - 생성된 이미지가 기준을 충족하는지 자동 검사한다."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import List
import numpy as np
try:
import cv2
from PIL import Image
except ImportError:
raise ImportError("opencv-python, Pillow이 필요합니다: pip install opencv-python Pillow")
@dataclass
class ValidationResult:
passed: bool
score: float # 0.0 ~ 1.0 종합 품질 점수
resolution_ok: bool
sharpness_ok: bool
color_diversity_ok: bool
messages: List[str]
@property
def summary(self) -> str:
status = "PASS" if self.passed else "FAIL"
return f"[{status}] score={self.score:.2f} | " + " | ".join(self.messages)
class QualityValidator:
"""이미지 품질을 자동으로 검증한다."""
def __init__(
self,
min_resolution: int = 2048,
sharpness_threshold: float = 100.0,
color_diversity_threshold: float = 0.15,
):
self.min_resolution = min_resolution
self.sharpness_threshold = sharpness_threshold
self.color_diversity_threshold = color_diversity_threshold
def validate(self, image_path: Path) -> ValidationResult:
if not image_path.exists():
return ValidationResult(
passed=False, score=0.0,
resolution_ok=False, sharpness_ok=False, color_diversity_ok=False,
messages=["파일이 존재하지 않음"],
)
img_pil = Image.open(image_path).convert("RGB")
img_cv = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)
resolution_ok, res_msg = self._check_resolution(img_pil)
sharpness_ok, sharp_score, sharp_msg = self._check_sharpness(img_cv)
color_ok, color_msg = self._check_color_diversity(img_cv)
scores = [
1.0 if resolution_ok else 0.0,
min(sharp_score / (self.sharpness_threshold * 3), 1.0),
1.0 if color_ok else 0.3,
]
overall_score = float(np.mean(scores))
passed = resolution_ok and sharpness_ok and color_ok
return ValidationResult(
passed=passed,
score=overall_score,
resolution_ok=resolution_ok,
sharpness_ok=sharpness_ok,
color_diversity_ok=color_ok,
messages=[res_msg, sharp_msg, color_msg],
)
def _check_resolution(self, img: Image.Image):
w, h = img.size
ok = w >= self.min_resolution and h >= self.min_resolution
msg = f"해상도={w}x{h} {'OK' if ok else f'(최소 {self.min_resolution} 필요)'}"
return ok, msg
def _check_sharpness(self, img_cv):
gray = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY)
lap_var = cv2.Laplacian(gray, cv2.CV_64F).var()
ok = lap_var >= self.sharpness_threshold
msg = f"선명도={lap_var:.1f} {'OK' if ok else f'(임계값 {self.sharpness_threshold})'}"
return ok, float(lap_var), msg
def _check_color_diversity(self, img_cv):
"""색상 다양성 검사 - 단색 평면 출력을 탐지한다."""
hsv = cv2.cvtColor(img_cv, cv2.COLOR_BGR2HSV)
s_channel = hsv[:, :, 1].astype(np.float32) / 255.0
mean_saturation = float(s_channel.mean())
ok = mean_saturation >= self.color_diversity_threshold
msg = f"색상다양성={mean_saturation:.3f} {'OK' if ok else f'(임계값 {self.color_diversity_threshold})'}"
return ok, msg

View File

@@ -0,0 +1,66 @@
"""Seed 관리자 - 작업별 Seed 고정 및 추적."""
from __future__ import annotations
import hashlib
import random
from typing import Optional
from sqlalchemy.orm import Session
from harness.logger import JobRecord, get_db_session
class SeedManager:
"""DXF 파일 해시 기반 결정론적 seed를 생성하고 이력을 관리한다."""
MAX_SEED = 2**32 - 1
def get_seed(
self,
file_hash: str,
fixed_seed: Optional[int] = None,
deterministic: bool = True,
) -> int:
"""
Args:
file_hash: DXF 파일의 SHA256 해시 앞 16자
fixed_seed: 사용자가 직접 지정한 seed (None이면 자동)
deterministic: True면 파일 해시 기반, False면 랜덤
"""
if fixed_seed is not None:
return int(fixed_seed) % (self.MAX_SEED + 1)
if deterministic:
return self._hash_to_seed(file_hash)
return random.randint(0, self.MAX_SEED)
def get_or_create_seed(
self,
db: Session,
job_id: int,
file_hash: str,
fixed_seed: Optional[int] = None,
deterministic: bool = True,
) -> int:
"""DB에서 기존 seed를 조회하거나 새로 생성한다."""
existing = db.query(JobRecord).filter_by(id=job_id).first()
if existing and existing.seed is not None:
return existing.seed
seed = self.get_seed(file_hash, fixed_seed, deterministic)
if existing:
existing.seed = seed
db.commit()
return seed
@staticmethod
def _hash_to_seed(file_hash: str) -> int:
"""파일 해시를 정수 seed로 변환한다."""
digest = hashlib.sha256(file_hash.encode()).digest()
return int.from_bytes(digest[:4], "big")
@staticmethod
def describe(seed: int) -> str:
return f"seed={seed} (0x{seed:08X})"