238 lines
8.5 KiB
Python
238 lines
8.5 KiB
Python
# 로그 시스템 — 앱 로그(텍스트) + 검사 결과(CSV) + 학습 이력(텍스트)
|
|
# 폴더 구조: logs/<category>/<YYYY-MM-DD>.<ext>
|
|
import builtins
|
|
import csv
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from paths import PROJECT_ROOT
|
|
|
|
LOGS_ROOT = Path(PROJECT_ROOT) / "logs"
|
|
_BORDER = "=" * 60
|
|
|
|
_orig_print = builtins.print
|
|
_session_start: datetime | None = None
|
|
_ACTION_LINE = "─" * 64
|
|
|
|
|
|
# ───────────────────────────── 경로 헬퍼 ───────────────────────────── #
|
|
|
|
def _date_path(category: str, ext: str = "log") -> Path:
|
|
"""logs/<category>/<YYYY-MM-DD>.<ext> 경로 (폴더 자동 생성)."""
|
|
now = datetime.now()
|
|
folder = LOGS_ROOT / category
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
return folder / f"{now:%Y-%m-%d}.{ext}"
|
|
|
|
|
|
def _append_text(path: Path, line: str):
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
if not line.endswith("\n"):
|
|
line += "\n"
|
|
f.write(line)
|
|
|
|
|
|
def _safe_write_app(line: str):
|
|
try:
|
|
_append_text(_date_path("app"), line)
|
|
except Exception as e:
|
|
_orig_print(f"[logger] 파일 쓰기 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
# ───────────────────────────── 세션 헤더 ───────────────────────────── #
|
|
|
|
def _previous_session_crashed() -> bool:
|
|
"""오늘 로그 파일에서 마지막 '세션 시작' 이후 '세션 종료'가 없으면 크래시로 간주."""
|
|
today = _date_path("app")
|
|
if not today.exists():
|
|
return False
|
|
try:
|
|
with open(today, "rb") as f:
|
|
try:
|
|
f.seek(-4096, os.SEEK_END)
|
|
except OSError:
|
|
f.seek(0)
|
|
tail = f.read().decode("utf-8", errors="ignore")
|
|
last_start = tail.rfind("=== 세션 시작")
|
|
last_end = tail.rfind("=== 세션 종료")
|
|
return last_start > last_end
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def setup_logging():
|
|
"""앱 시작 시 호출. print 가로채기 + 세션 시작 헤더 기록."""
|
|
global _session_start
|
|
|
|
crashed = _previous_session_crashed()
|
|
_session_start = datetime.now()
|
|
pid = os.getpid()
|
|
|
|
header = [
|
|
"",
|
|
_BORDER,
|
|
f"=== 세션 시작 : {_session_start:%Y-%m-%d %H:%M:%S} PID {pid}",
|
|
]
|
|
if crashed:
|
|
header.append("=== ⚠ 직전 세션 비정상 종료 감지 ===")
|
|
header.append(_BORDER)
|
|
for line in header:
|
|
_safe_write_app(line)
|
|
|
|
def _ts_print(*args, sep=" ", end="\n", file=None, flush=False, **kwargs):
|
|
now = datetime.now()
|
|
msg = sep.join(str(a) for a in args)
|
|
# 콘솔: [HH:MM:SS]
|
|
_orig_print(
|
|
f"[{now:%H:%M:%S}]", *args,
|
|
sep=sep, end=end, file=file, flush=flush, **kwargs,
|
|
)
|
|
# 파일: 풀 시각 + 메시지
|
|
_safe_write_app(f"{now:%Y-%m-%d %H:%M:%S.%f}"[:-3] + f" {msg}")
|
|
|
|
builtins.print = _ts_print
|
|
|
|
|
|
def teardown_logging():
|
|
"""앱 정상 종료 시 호출. 세션 종료 헤더 기록."""
|
|
global _session_start
|
|
if _session_start is None:
|
|
return
|
|
end = datetime.now()
|
|
secs = int((end - _session_start).total_seconds())
|
|
h, rem = divmod(secs, 3600)
|
|
m, s = divmod(rem, 60)
|
|
|
|
for line in [
|
|
_BORDER,
|
|
f"=== 세션 종료 : {end:%Y-%m-%d %H:%M:%S} 운영시간 {h:02d}:{m:02d}:{s:02d}",
|
|
_BORDER,
|
|
"",
|
|
]:
|
|
_safe_write_app(line)
|
|
_session_start = None
|
|
|
|
|
|
# ───────────────────────────── 사용자 액션 ───────────────────────────── #
|
|
|
|
def log_action(msg: str):
|
|
"""버튼 클릭·기능 실행 등 사용자 액션을 구분선으로 강조해 app 로그에 기록."""
|
|
_safe_write_app(_ACTION_LINE)
|
|
print(f"▶ {msg}")
|
|
_safe_write_app(_ACTION_LINE)
|
|
|
|
|
|
# ───────────────────────────── 검사 결과 CSV ───────────────────────────── #
|
|
|
|
_INSPECT_HEADER = [
|
|
"timestamp", "group", "result",
|
|
"cognex_pass", "basler_pass", "detected_models",
|
|
]
|
|
|
|
|
|
def log_inspect_result(
|
|
group: str,
|
|
result: str,
|
|
cognex_pass: bool | None = None,
|
|
basler_pass: bool | None = None,
|
|
detected: list[str] | None = None,
|
|
):
|
|
"""검사 결과 1건을 logs/inspect/YYYY-MM-DD.csv 에 append.
|
|
파일 신규일 때만 헤더 행 작성. utf-8-sig 로 Excel 호환."""
|
|
csv_path = _date_path("inspect", ext="csv")
|
|
is_new = not csv_path.exists()
|
|
try:
|
|
with open(csv_path, "a", encoding="utf-8-sig", newline="") as f:
|
|
w = csv.writer(f)
|
|
if is_new:
|
|
w.writerow(_INSPECT_HEADER)
|
|
w.writerow([
|
|
f"{datetime.now():%Y-%m-%d %H:%M:%S}",
|
|
group,
|
|
result,
|
|
"" if cognex_pass is None else ("Y" if cognex_pass else "N"),
|
|
"" if basler_pass is None else ("Y" if basler_pass else "N"),
|
|
",".join(detected) if detected else "",
|
|
])
|
|
except Exception as e:
|
|
_orig_print(f"[logger] 검사 CSV 기록 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
# ───────────────────────────── 카메라 타이밍 CSV ───────────────────────────── #
|
|
|
|
_TIMING_HEADER = ["timestamp", "seq", "event", "elapsed_ms", "detail"]
|
|
_timing_lock = __import__("threading").Lock()
|
|
|
|
|
|
def log_camera_timing(seq: int, event: str, elapsed_ms: float, detail: str = ""):
|
|
"""카메라 촬영 타이밍 1이벤트를 logs/timing/YYYY-MM-DD.csv 에 append.
|
|
Cognex/Basler 서브스레드에서 호출되므로 Lock으로 동시 쓰기 보호."""
|
|
csv_path = _date_path("timing", ext="csv")
|
|
is_new = not csv_path.exists()
|
|
try:
|
|
with _timing_lock:
|
|
with open(csv_path, "a", encoding="utf-8-sig", newline="") as f:
|
|
w = csv.writer(f)
|
|
if is_new:
|
|
w.writerow(_TIMING_HEADER)
|
|
w.writerow([
|
|
f"{datetime.now():%H:%M:%S.%f}"[:-3],
|
|
seq,
|
|
event,
|
|
f"{elapsed_ms:.1f}",
|
|
detail,
|
|
])
|
|
except Exception as e:
|
|
_orig_print(f"[logger] 타이밍 CSV 기록 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
# ───────────────────────────── 학습 로그 ───────────────────────────── #
|
|
|
|
def log_train(message: str):
|
|
"""재학습 이벤트 1건을 logs/train/YYYY/MM/YYYY-MM-DD.log 에 append."""
|
|
line = f"{datetime.now():%Y-%m-%d %H:%M:%S} {message}"
|
|
try:
|
|
_append_text(_date_path("train"), line)
|
|
except Exception as e:
|
|
_orig_print(f"[logger] 학습 로그 기록 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
# ───────────────────────────── 불량 이미지 저장 ───────────────────────────── #
|
|
|
|
def log_defect_image(image: "np.ndarray", defects: list) -> bool:
|
|
"""불량 감지 이미지를 logs/defect/YYYY-MM-DD/HHMMSS_SSS_{클래스}.jpg 에 저장.
|
|
반환: 저장 성공 여부.
|
|
"""
|
|
import cv2
|
|
import numpy as np # noqa: F401 — type annotation용
|
|
|
|
now = datetime.now()
|
|
folder = LOGS_ROOT / "defect" / now.strftime("%Y-%m-%d")
|
|
try:
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
except Exception as e:
|
|
print(f"[logger] 불량 이미지 폴더 생성 실패: {e} 경로: {folder}")
|
|
return False
|
|
|
|
# 중복 제거 후 순서 유지한 클래스명 목록
|
|
seen = set()
|
|
names = [
|
|
d["class_name"] for d in defects
|
|
if not (d["class_name"] in seen or seen.add(d["class_name"]))
|
|
]
|
|
ms = now.microsecond // 1000
|
|
filename = f"{now:%H%M%S}_{ms:03d}_{'_'.join(names)}.jpg"
|
|
path = folder / filename
|
|
|
|
try:
|
|
ok = cv2.imwrite(str(path), image)
|
|
if not ok:
|
|
print(f"[logger] 불량 이미지 저장 실패 (imwrite 반환 False): {path}")
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
print(f"[logger] 불량 이미지 저장 예외: {e} 경로: {path}")
|
|
return False
|