# 검사 페이지 — 코그넥스/Basler 영상, 그룹 A/B 설정, Pass/Fail 표시 import time import threading import itertools import cv2 import numpy as np from PyQt5.QtCore import Qt, QThread, pyqtSignal, QSize from PyQt5.QtGui import QImage, QPixmap from PyQt5.QtWidgets import ( QWidget, QHBoxLayout, QVBoxLayout, QGroupBox, QPushButton, QLabel, QCheckBox, QFrame, QGridLayout, QSizePolicy, ) from logic.inspector import Inspector from logic.group_manager import GroupManager from logger import log_inspect_result, log_camera_timing, log_action, log_defect_image _DEFECT_COLORS = { "스크래치": (0, 0, 255), "이물": (0, 165, 255), "흑점": (128, 0, 128), "변형": (255, 165, 0 ), } def _draw_detections(frame: np.ndarray, detections: list) -> np.ndarray: """frame 복사본에 BBox 오버레이를 그려 반환.""" img = frame.copy() for det in detections: x1, y1, x2, y2 = [int(v) for v in det["bbox"]] name = det["class_name"] conf = det["confidence"] color = _DEFECT_COLORS.get(name, (0, 255, 0)) cv2.rectangle(img, (x1, y1), (x2, y2), color, 2) cv2.putText( img, f"{name} {conf:.0%}", (x1, max(y1 - 8, 0)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, ) return img # 검사 그룹 A/B 선택용 모델 목록 _MODELS = [ "LOW REF / LX3 / RH", "LOW REF / LX3 / LH", "LOW REF NAS / LX3 / RH", "LOW REF NAS / LX3 / LH", "LOW REF NAS / MX5a 2.0TH / RH", "LOW REF NAS / MX5a 2.0TH / LH", "HIGH REF / LX3 / RH", "HIGH REF / LX3 / LH", "LOW REF NAS 1.5 GEN / CN7 PE / RH", "LOW REF DOM 1.5 GEN / CN7 PE / LH", ] _MODEL_ID_MAP = { "LOW REF / LX3 / RH": 1, "LOW REF / LX3 / LH": 2, "LOW REF NAS / LX3 / RH": 3, "LOW REF NAS / LX3 / LH": 4, "LOW REF NAS / MX5a 2.0TH / RH": 5, "LOW REF NAS / MX5a 2.0TH / LH": 6, "HIGH REF / LX3 / RH": 7, "HIGH REF / LX3 / LH": 8, "LOW REF NAS 1.5 GEN / CN7 PE / RH": 9, "LOW REF DOM 1.5 GEN / CN7 PE / LH": 10, } # ================================================================== # # 백그라운드 워커 — 파이프라인 방식 # # [Cognex 서브스레드] trigger → sleep(1.0) → FTP(영구세션) → PatMax # ↕ 병렬 ↕ join # [워커 메인] sleep(belt_delay) → Basler 캡처 → 판정 → emit # # belt_delay = 카메라 간 거리(cm) / 벨트 속도(cm/s) # 두 작업이 동시에 시작되어 같은 제품을 각 위치에서 촬영함. # ================================================================== # class InspectWorker(QThread): cognex_image_ready = pyqtSignal(bytes) # raw BMP/JPG 바이트 basler_image_ready = pyqtSignal(object, list) # (ndarray, detections) result_ready = pyqtSignal(dict) def __init__(self, insight, basler, inspector, groups, belt_delay: float = 3.33, parent=None): super().__init__(parent) self._insight = insight self._basler = basler self._inspector = inspector self._groups = groups self.detector = None self.matcher = None # PatternMatcher — InspectPage에서 주입 self._belt_delay = belt_delay self._stop_flag = False self._pause_flag = False self._seq = itertools.count(1) # ── 외부 제어 ──────────────────────────────────────────────────── # def stop(self): self._stop_flag = True def pause(self): self._pause_flag = True def resume(self): self._pause_flag = False def set_belt_delay(self, delay: float): self._belt_delay = delay # ── 스레드 본체 ────────────────────────────────────────────────── # def run(self): self._stop_flag = False self._pause_flag = False while not self._stop_flag: if self._pause_flag: time.sleep(0.1) continue self._do_one_cycle() # ── 검사 1사이클 ───────────────────────────────────────────────── # def _do_one_cycle(self): group = self._groups.get_active_name() seq = next(self._seq) trigger_time = time.perf_counter() def _ms() -> float: return (time.perf_counter() - trigger_time) * 1000 log_camera_timing(seq, "cycle_start", 0.0, f"group={group} belt_delay={self._belt_delay:.2f}s") # ── Cognex 작업: 서브 스레드에서 병렬 실행 ── cognex_out: dict = {} def _cognex_work(): try: log_camera_timing(seq, "cognex_trigger_send", _ms()) ok = self._insight.software_trigger() log_camera_timing(seq, f"cognex_trigger_{'ok' if ok else 'fail'}", _ms()) if not ok: cognex_out["error"] = "trigger_failed" return time.sleep(1.0) log_camera_timing(seq, "cognex_ftp_start", _ms()) raw = self._insight.get_image() log_camera_timing( seq, "cognex_ftp_done", _ms(), f"{len(raw)}bytes" if raw else "empty", ) if raw: self.cognex_image_ready.emit(raw) log_camera_timing(seq, "cognex_patmax_start", _ms()) # Cognex job 파일 결과 (항상) gv_results = self._inspector.read_patmax_results(self._insight) # Python ORB 결과 (추가 등록 제품, 있을 때만) py_results = {} if self.matcher and self.matcher.registered_ids and raw: py_results = self._inspector.match_image(raw, self.matcher) cognex_out["results"] = {**gv_results, **py_results} log_camera_timing(seq, "cognex_patmax_done", _ms()) except Exception as e: print(f"[워커] Cognex 서브스레드 오류: {e}") cognex_out["error"] = str(e) log_camera_timing(seq, "cognex_error", _ms(), str(e)) ct = threading.Thread(target=_cognex_work, daemon=True) ct.start() # ── Basler: trigger 시점 기준 belt_delay 후 캡처 ── elapsed = time.perf_counter() - trigger_time remaining = self._belt_delay - elapsed if remaining > 0: time.sleep(remaining) basler_pass = True basler_detections = [] try: log_camera_timing(seq, "basler_capture_start", _ms()) frame = self._basler.capture() log_camera_timing( seq, "basler_capture_done", _ms(), f"{frame.shape}" if frame is not None else "failed", ) if frame is not None: if self.detector and self.detector.is_loaded(): detections = self.detector.detect(frame) defects = [d for d in detections if d["confidence"] >= 0.5] basler_pass = len(defects) == 0 basler_detections = defects if defects: print(f"[워커] 불량 감지: {[d['class_name'] for d in defects]}") annotated = _draw_detections(frame, defects) log_defect_image(annotated, defects) self.basler_image_ready.emit(frame, basler_detections) except Exception as e: print(f"[워커 오류] Basler: {e}") log_camera_timing(seq, "basler_error", _ms(), str(e)) # ── Cognex 서브스레드 완료 대기 ── log_camera_timing(seq, "cognex_join_wait", _ms()) ct.join(timeout=10.0) log_camera_timing(seq, "cognex_join_done", _ms()) # ── 모델 판별 ── results = cognex_out.get("results", {}) active_names = self._groups.get_active_group() allowed_ids = [_MODEL_ID_MAP[n] for n in active_names if n in _MODEL_ID_MAP] result_info = { "matched": False, "in_allowed": False, "model": None, "score": 0.0, "cognex_pass": False, "status": "인식 불가", } try: if results: result_info = self._inspector.identify_model(results, allowed_ids) except Exception as e: print(f"[워커 오류] 모델 판별: {e}") cognex_pass = result_info["cognex_pass"] matched = result_info["matched"] final = self._inspector.judge(cognex_pass, basler_pass) self.result_ready.emit({ "group": group, "matched": matched, "result": final, "cognex_pass": cognex_pass, "basler_pass": basler_pass, "result_info": result_info, }) log_camera_timing( seq, "cycle_done", _ms(), f"result={final} cognex={'PASS' if cognex_pass else 'FAIL'} basler={'PASS' if basler_pass else 'FAIL'}", ) try: log_inspect_result( group=group, result=("UNKNOWN" if not matched else final), cognex_pass=cognex_pass, basler_pass=basler_pass, detected=[result_info["model"]] if result_info.get("model") else None, ) except Exception as e: print(f"[워커 오류] 로그 기록: {e}") # ================================================================== # # 검사 페이지 # ================================================================== # class InspectPage(QWidget): def __init__(self, insight_cam, basler_cam, detector=None, belt_delay: float = 3.33, parent=None): super().__init__(parent) self._insight = insight_cam self._basler = basler_cam self.detector = detector self._inspector = Inspector() self._groups = GroupManager() self._counts = { "A": {"total": 0, "pass": 0, "fail": 0, "unknown": 0}, "B": {"total": 0, "pass": 0, "fail": 0, "unknown": 0}, } self._matcher = None self._worker = InspectWorker( self._insight, self._basler, self._inspector, self._groups, belt_delay=belt_delay, ) self._worker.detector = self.detector self._worker.matcher = self._matcher self._worker.cognex_image_ready.connect(self._display_cognex_image) self._worker.basler_image_ready.connect(self._on_basler_ready) self._worker.result_ready.connect(self._on_result) self._build_ui() # ================================================================== # # 최상위 레이아웃 # ================================================================== # def _build_ui(self): root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) root.addWidget(self._build_top(), stretch=7) root.addWidget(self._build_bottom(), stretch=3) # ================================================================== # # 상단: 코그넥스 (좌 50 %) / Basler (우 50 %) # ================================================================== # def _build_top(self) -> QWidget: w = QWidget() w.setStyleSheet("background:#0d0d0d;") layout = QHBoxLayout(w) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) layout.addLayout(self._build_cam_col("■ In-Sight 2000C", "#4488ff", "_cognex_label"), stretch=1) layout.addWidget(_vline()) layout.addLayout(self._build_cam_col("■ Basler USB", "#44cc88", "_basler_label"), stretch=1) return w def _build_cam_col(self, title: str, color: str, attr: str) -> QVBoxLayout: col = QVBoxLayout() col.setContentsMargins(0, 0, 0, 0) col.setSpacing(0) title_lbl = QLabel(title) title_lbl.setFixedHeight(32) title_lbl.setStyleSheet( f"color:{color}; font-size:15px; font-weight:bold;" "padding-left:10px; background:#111111;" ) img_lbl = QLabel() img_lbl.setAlignment(Qt.AlignCenter) img_lbl.setStyleSheet("background:#0d0d0d;") img_lbl.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) img_lbl.setMinimumSize(640, 480) img_lbl.setScaledContents(False) setattr(self, attr, img_lbl) col.addWidget(title_lbl) col.addWidget(img_lbl, stretch=1) return col # ================================================================== # # 하단: 3열 # ================================================================== # def _build_bottom(self) -> QWidget: w = QWidget() w.setStyleSheet("background:#1a1a1a;") layout = QHBoxLayout(w) layout.setContentsMargins(8, 6, 8, 6) layout.setSpacing(0) layout.addWidget(self._build_col_groups(), stretch=5) layout.addWidget(_vline()) layout.addWidget(self._build_col_controls(), stretch=3) layout.addWidget(_vline()) layout.addWidget(self._build_col_counters(), stretch=4) return w def _build_col_groups(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) layout.setContentsMargins(4, 4, 8, 4) layout.setSpacing(6) checks_row = QHBoxLayout() checks_row.setSpacing(8) checks_row.addWidget(self._build_group_section("A"), stretch=1) checks_row.addWidget(self._build_group_section("B"), stretch=1) layout.addLayout(checks_row, stretch=1) self._switch_btn = QPushButton("현재: 그룹 A 활성 → B로 전환") self._switch_btn.setFixedHeight(56) self._switch_btn.setStyleSheet( "background:#1a3a5c; color:#ffffff; border:none; border-radius:4px;" "font-size:14px; font-weight:bold;" ) self._switch_btn.clicked.connect(self._on_switch) layout.addWidget(self._switch_btn) return w def _build_group_section(self, name: str) -> QGroupBox: active_color = "#4488ff" if name == "A" else "#cc8844" g = QGroupBox(f"그룹 {name} (최대 4종)") g.setStyleSheet( f"QGroupBox {{ background:#222222; border:1px solid #333333; border-radius:6px;" f" margin-top:12px; padding:6px 4px 4px 4px; }}" f"QGroupBox::title {{ color:{active_color}; subcontrol-origin:margin;" f" left:8px; font-size:13px; font-weight:bold; }}" ) layout = QVBoxLayout(g) layout.setSpacing(1) layout.setContentsMargins(4, 2, 4, 2) checks = [] for model in _MODELS: cb = QCheckBox(model) cb.setStyleSheet( "QCheckBox { font-size:12px; min-height:24px; color:#cccccc; }" "QCheckBox::indicator { width:16px; height:16px; }" ) checks.append(cb) layout.addWidget(cb) if name == "A": self._group_a_checks = checks for cb in checks: cb.clicked.connect(lambda checked, c=cb: self._on_group_changed("A", c, checked)) else: self._group_b_checks = checks for cb in checks: cb.clicked.connect(lambda checked, c=cb: self._on_group_changed("B", c, checked)) return g def _build_col_controls(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) layout.setContentsMargins(12, 4, 12, 4) layout.setSpacing(6) self._start_btn = QPushButton("검사 시작") self._start_btn.setFixedHeight(70) self._start_btn.setStyleSheet( "background:#1a5c1a; color:#ffffff; border:none; border-radius:4px;" "font-size:18px; font-weight:bold;" ) self._start_btn.clicked.connect(self._on_start) self._pause_btn = QPushButton("일시 정지") self._pause_btn.setFixedHeight(70) self._pause_btn.setEnabled(False) self._pause_btn.setStyleSheet( "background:#5c5500; color:#ffffff; border:none; border-radius:4px;" "font-size:18px; font-weight:bold;" ) self._pause_btn.clicked.connect(self._on_pause) self._active_lbl = QLabel("활성 그룹: A") self._active_lbl.setAlignment(Qt.AlignCenter) self._active_lbl.setStyleSheet("font-size:14px; color:#aaaaaa;") self._model_lbl = QLabel("인식 모델: —") self._model_lbl.setAlignment(Qt.AlignCenter) self._model_lbl.setWordWrap(True) self._model_lbl.setStyleSheet("font-size:13px; color:#cccccc;") self._belt_lbl = QLabel( f"벨트 딜레이: {self._worker._belt_delay:.2f}s" ) self._belt_lbl.setAlignment(Qt.AlignCenter) self._belt_lbl.setStyleSheet("font-size:12px; color:#666666;") self._result_lbl = QLabel("대기 중") self._result_lbl.setAlignment(Qt.AlignCenter) self._result_lbl.setStyleSheet( "font-size:48px; font-weight:bold; background:#2a2a2a;" "color:#666666; border-radius:8px;" ) self._result_lbl.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) layout.addWidget(self._start_btn) layout.addWidget(self._pause_btn) layout.addWidget(self._active_lbl) layout.addWidget(self._model_lbl) layout.addWidget(self._belt_lbl) layout.addWidget(self._result_lbl, stretch=1) return w def _build_col_counters(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) layout.setContentsMargins(12, 4, 8, 4) layout.setSpacing(6) grid = QGridLayout() grid.setSpacing(4) for col, text in enumerate(["", "그룹 A", "그룹 B"]): lbl = QLabel(text) lbl.setAlignment(Qt.AlignCenter) lbl.setStyleSheet("font-size:13px; color:#888888; font-weight:bold;") grid.addWidget(lbl, 0, col) row_defs = [ ("전체", "total", "#ffffff"), ("양품", "pass", "#22cc55"), ("불량", "fail", "#cc2222"), ("미인식", "unknown", "#ff9900"), ] self._cnt_lbls = {} for r, (display, key, color) in enumerate(row_defs, start=1): row_lbl = QLabel(display) row_lbl.setStyleSheet(f"font-size:14px; color:{color}; font-weight:bold;") grid.addWidget(row_lbl, r, 0) self._cnt_lbls[key] = {} for g_col, group in enumerate(["A", "B"], start=1): lbl = QLabel("0") lbl.setAlignment(Qt.AlignCenter) lbl.setStyleSheet( f"font-size:36px; font-weight:bold; color:{color};" "background:#222222; border-radius:4px; padding:2px 6px;" ) grid.addWidget(lbl, r, g_col) self._cnt_lbls[key][group] = lbl layout.addLayout(grid) btn_reset = QPushButton("카운터 초기화") btn_reset.setFixedHeight(56) btn_reset.setStyleSheet( "background:#3a1a1a; color:#ffffff; border:none; border-radius:4px; font-size:14px;" ) btn_reset.clicked.connect(self._on_reset) layout.addWidget(btn_reset) layout.addStretch() return w # ================================================================== # # 슬롯 — UI 이벤트 # ================================================================== # def _on_group_changed(self, group: str, changed_cb: QCheckBox, is_checked: bool): checks = self._group_a_checks if group == "A" else self._group_b_checks if is_checked and sum(1 for c in checks if c.isChecked()) > GroupManager.MAX_PER_GROUP: changed_cb.setChecked(False) return models = [c.text() for c in checks if c.isChecked()] if group == "A": self._groups.set_group_a(models) else: self._groups.set_group_b(models) def _on_switch(self): active = self._groups.switch_group() other = "B" if active == "A" else "A" self._switch_btn.setText(f"현재: 그룹 {active} 활성 → {other}로 전환") self._active_lbl.setText(f"활성 그룹: {active}") print(f"[검사] 그룹 전환 → 활성 그룹 {active}") def _on_start(self): if self._worker.isRunning(): return log_action("[검사] 검사 시작") self._start_btn.setEnabled(False) self._pause_btn.setEnabled(True) self._pause_btn.setText("일시 정지") self._worker.start() def _on_pause(self): if self._worker.isRunning() and not self._worker._pause_flag: self._worker.pause() self._pause_btn.setText("재개") self._start_btn.setEnabled(True) log_action("[검사] 일시 정지") else: self._worker.resume() self._pause_btn.setText("일시 정지") self._start_btn.setEnabled(False) log_action("[검사] 검사 재개") def _on_reset(self): log_action("[검사] 카운트 리셋") for key in ("total", "pass", "fail", "unknown"): for g in ("A", "B"): self._counts[g][key] = 0 self._cnt_lbls[key][g].setText("0") def closeEvent(self, event): self._worker.stop() self._worker.wait(3000) super().closeEvent(event) # ================================================================== # # 외부에서 호출하는 업데이트 메서드 # ================================================================== # def update_detector(self, detector): self.detector = detector self._worker.detector = detector print(f"[검사] AI 모델 업데이트: {detector.model_path if detector else None}") def update_matcher(self, matcher): self._matcher = matcher self._worker.matcher = matcher n = len(matcher.registered_ids) if matcher else 0 print(f"[검사] PatternMatcher 업데이트: {n}개 패턴") def update_insight(self, new_insight): """카메라 재연결 시 워커에도 반영.""" self._insight = new_insight self._worker._insight = new_insight def update_basler(self, new_basler): self._basler = new_basler self._worker._basler = new_basler def update_belt_delay(self, delay: float): """설정 페이지에서 컨베이어 값 변경 시 호출.""" self._worker.set_belt_delay(delay) self._belt_lbl.setText(f"벨트 딜레이: {delay:.2f}s") # ================================================================== # # 워커 signal 슬롯 (메인 스레드) # ================================================================== # def _on_basler_ready(self, frame, detections): self._display_basler_image(frame, detections=detections) def _on_result(self, data: dict): group = data["group"] matched = data["matched"] result = data["result"] cognex_pass = data["cognex_pass"] basler_pass = data["basler_pass"] result_info = data["result_info"] self._model_lbl.setText(result_info["status"]) self._counts[group]["total"] += 1 if not matched: self._counts[group]["unknown"] += 1 elif result == "PASS": self._counts[group]["pass"] += 1 else: self._counts[group]["fail"] += 1 for key in ("total", "pass", "fail", "unknown"): self._cnt_lbls[key][group].setText(str(self._counts[group][key])) if not matched: self._set_result("미인식", "#332200", "#ff9900") elif result == "PASS": self._set_result("PASS", "#003300", "#22ff55") else: self._set_result("FAIL", "#330000", "#ff2222") # ================================================================== # # 이미지 표시 헬퍼 # ================================================================== # def _set_result(self, text: str, bg: str, fg: str): self._result_lbl.setText(text) self._result_lbl.setStyleSheet( f"font-size:48px; font-weight:bold; background:{bg};" f"color:{fg}; border-radius:8px;" ) def _display_cognex_image(self, raw_data: bytes): try: arr = np.frombuffer(raw_data, dtype=np.uint8) img = cv2.imdecode(arr, cv2.IMREAD_COLOR) if img is None: img = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED) if img is None: print("[코그넥스] 이미지 디코딩 실패") return if len(img.shape) == 2: h, w = img.shape qimg = QImage(img.data.tobytes(), w, h, w, QImage.Format_Grayscale8) else: rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) h, w, ch = rgb.shape qimg = QImage(rgb.data.tobytes(), w, h, w * ch, QImage.Format_RGB888) self._cognex_label.setPixmap( QPixmap.fromImage(qimg).scaled( self._cognex_label.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) except Exception as e: print(f"[코그넥스] 이미지 표시 오류: {e}") def _display_basler_image(self, frame, detections=None): try: img = _draw_detections(frame, detections or []) if len(img.shape) == 2: h, w = img.shape qimg = QImage(img.data.tobytes(), w, h, w, QImage.Format_Grayscale8) else: rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) h, w, ch = rgb.shape qimg = QImage(rgb.data.tobytes(), w, h, w * ch, QImage.Format_RGB888) self._basler_label.setPixmap( QPixmap.fromImage(qimg).scaled( self._basler_label.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) except Exception as e: print(f"[Basler] 이미지 표시 오류: {e}") # ================================================================== # # 모듈 수준 유틸리티 # ================================================================== # def _vline() -> QFrame: f = QFrame() f.setFrameShape(QFrame.VLine) f.setFixedWidth(2) f.setStyleSheet("background:#333333; border:none;") return f def _raw_to_pixmap(raw: bytes, size: QSize) -> "QPixmap | None": arr = np.frombuffer(raw, dtype=np.uint8) img = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED) if img is None: return None return _ndarray_to_pixmap(img, size) def _ndarray_to_pixmap(img: np.ndarray, size: QSize) -> "QPixmap | None": if img.ndim == 2: img_c = np.ascontiguousarray(img) h, w = img_c.shape qimg = QImage(img_c.data, w, h, w, QImage.Format_Grayscale8) else: rgb = np.ascontiguousarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB)) h, w, ch = rgb.shape qimg = QImage(rgb.data, w, h, w * ch, QImage.Format_RGB888) return QPixmap.fromImage(qimg).scaled(size, Qt.KeepAspectRatio, Qt.SmoothTransformation)