diff --git a/README.md b/README.md
index ea7ca98..4e814e5 100644
--- a/README.md
+++ b/README.md
@@ -1 +1,59 @@
-# Smartfactory
\ No newline at end of file
+
+
+# Intel Geti + AI QC Conveyor
+인텔 Geti + 인공지능 QC 컨베이어
+
+
+## 프로젝트 소개
+인건비 절감을 위한 생산 공장 컨베이어 자동화 장치 입니다.
+
+
+## 개발 기간
+* 25.09.24 - 25.10.22
+
+### 맴버구성
+| 이름 | 담당 |
+|------|------|
+|서채건|PM, GUI 구성, 코드 통합|
+|김동현|코드 통합, PPT제작, 발표|
+|김선곤|GUI 구성, DB 읽기 쓰기 로직|
+|박서정|데이터셋 학습 및 검증, 스티커 유무 감별|
+|조민재|데이터셋 학습 및 검증, 오염 비율 계산|
+
+### 개발 환경
+- `Intel Geti`
+- `Python 3.12.3`
+- `Arduino Mega`
+- `MariaDB 10`
+- `Tkinter`
+
+## 주요 기능
+#### 모델 1 작동
+- Pink 병뚜껑 QC 스티커 유무, 오염 유무
+- No 스티커 or 오염 30% 초과 or Purple 병뚜껑 -> 완전불량
+- 스티커 and 오염 0% 초과 30% 이하 -> 부분불량
+- 스티커 and 오염 0% -> 정상
+
+#### 모델 2 작동
+- Purple 병뚜껑 QC 스티커 유무, 오염 유무
+- No 스티커 or 오염 30% 초과 or Pink 병뚜껑 -> 완전불량
+- 스티커 and 오염 0% 초과 30% 이하 -> 부분불량
+- 스티커 and 오염 0% -> 정상
+
+#### 관리자 모드
+- 수동 컨베이어 제어
+- 수동 엑츄에이터 조작
+
+## 파일 구조
+```
+team2/
+├─ iotdemo/
+│ ├─ __init__.py
+│ ├─ debounce.py
+│ ├─ factory_controller.py
+│ ├─ pins.py
+│ ├─ pyduino.py
+│ └─ pyft232.py
+├─ MODEL_FILE
+└─ run.py
+```
diff --git a/arduino/arduino.ino b/arduino/arduino.ino
new file mode 100644
index 0000000..aa5ea1d
--- /dev/null
+++ b/arduino/arduino.ino
@@ -0,0 +1,148 @@
+/* Packet Processing */
+#define PACKET_START (0x55)
+#define PACKET_END (0xaa)
+
+#define PACKET_ID_INTERRUPT (0x00)
+#define PACKET_ID_MAX (199)
+#define PACKET_ID_ERROR (0xff)
+
+static byte packet[5] = { PACKET_START, 0xff, 0xff, 0xff, PACKET_END };
+
+enum pin_functions
+{
+ BEACON_RED = 2,
+ BEACON_ORANGE,
+ BEACON_GREEN,
+ BEACON_BUZZER,
+
+ ACTUATOR_1, // 6
+ ACTUATOR_2,
+ CONVEYOR_EN,
+ CONVEYOR_PWM,
+
+ START_BUTTON, // 10
+ STOP_BUTTON,
+ PHOTOELECTRIC_SENSOR_1,
+ PHOTOELECTRIC_SENSOR_2,
+
+ LED // 14
+};
+
+#define SEND_PACKET(pin, value) \
+ do \
+ { \
+ packet[2] = (pin); \
+ packet[3] = (value); \
+ Serial.write(packet, sizeof(packet)); \
+ } while (0)
+
+/* Utility Functions */
+#define INIT_OUTPUT_PAD(pad, val) \
+ pinMode((pad), OUTPUT); \
+ digitalWrite((pad), (val))
+
+#define INIT_PWM_PAD(pad, val) \
+ pinMode((pad), OUTPUT); \
+ analogWrite((pad), (val))
+
+#define CHECK(expr) \
+ if ((expr)) \
+ { \
+ SEND_PACKET(0xff, 0xff); \
+ return; \
+ }
+
+/* System Initialization */
+void setup()
+{
+ /* Initialize Outputs */
+ INIT_OUTPUT_PAD(BEACON_RED, HIGH);
+ INIT_OUTPUT_PAD(BEACON_ORANGE, HIGH);
+ INIT_OUTPUT_PAD(BEACON_GREEN, HIGH);
+
+ INIT_OUTPUT_PAD(BEACON_BUZZER, HIGH);
+
+ INIT_OUTPUT_PAD(ACTUATOR_1, HIGH);
+ INIT_OUTPUT_PAD(ACTUATOR_2, HIGH);
+
+ INIT_OUTPUT_PAD(CONVEYOR_EN, HIGH);
+ INIT_PWM_PAD(CONVEYOR_PWM, 0);
+
+ INIT_OUTPUT_PAD(LED, HIGH);
+
+ /* Initialize Inputs */
+ pinMode(START_BUTTON, INPUT_PULLUP);
+ pinMode(STOP_BUTTON, INPUT_PULLUP);
+ pinMode(PHOTOELECTRIC_SENSOR_1, INPUT_PULLUP);
+ pinMode(PHOTOELECTRIC_SENSOR_2, INPUT_PULLUP);
+
+ /* Prepare UART port - 115200 baud rate */
+ Serial.begin(115200);
+ while (!Serial)
+ ;
+}
+
+/* Set to 0xff to report initial state */
+static byte inputStatus[4] = { 0xff, 0xff, 0xff, 0xff };
+
+static inline compareAndReport(byte pin)
+{
+ static byte idx;
+ static byte value;
+
+ idx = pin - START_BUTTON;
+ value = digitalRead(pin);
+ if (inputStatus[idx] == value)
+ return;
+
+ inputStatus[idx] = value;
+ SEND_PACKET(pin, value);
+}
+
+static inline void polling(void)
+{
+ packet[1] = PACKET_ID_INTERRUPT;
+
+ compareAndReport(START_BUTTON);
+ compareAndReport(STOP_BUTTON);
+ compareAndReport(PHOTOELECTRIC_SENSOR_1);
+ compareAndReport(PHOTOELECTRIC_SENSOR_2);
+}
+
+/* Actual Main loop */
+void loop()
+{
+ polling();
+
+ /* Packet Processing */
+ if (Serial.available() < 5)
+ return;
+
+ static byte packetId;
+ static byte pin;
+ static byte value;
+
+ packet[1] = PACKET_ID_ERROR;
+
+ CHECK(Serial.read() != PACKET_START);
+
+ packetId = Serial.read();
+ CHECK(packetId == 0 || packetId > PACKET_ID_MAX);
+
+ pin = Serial.read();
+ CHECK(pin < BEACON_RED || pin > LED);
+
+ value = Serial.read();
+
+ CHECK(Serial.read() != PACKET_END);
+
+ /* Run command */
+ packet[1] = packetId;
+
+ if (pin == CONVEYOR_PWM)
+ analogWrite(pin, value);
+ else
+ digitalWrite(pin, !!value);
+
+ SEND_PACKET(pin, value);
+}
diff --git a/class.onnx b/class.onnx
new file mode 100644
index 0000000..b011b76
Binary files /dev/null and b/class.onnx differ
diff --git a/det.onnx b/det.onnx
new file mode 100644
index 0000000..db664ff
Binary files /dev/null and b/det.onnx differ
diff --git a/doc/Team2_1Pager.png b/doc/Team2_1Pager.png
new file mode 100644
index 0000000..90c4be4
Binary files /dev/null and b/doc/Team2_1Pager.png differ
diff --git a/doc/Team2_발표_PDF.pdf b/doc/Team2_발표_PDF.pdf
new file mode 100644
index 0000000..ce9c786
Binary files /dev/null and b/doc/Team2_발표_PDF.pdf differ
diff --git a/iotdemo/__init__.py b/iotdemo/__init__.py
new file mode 100644
index 0000000..b887938
--- /dev/null
+++ b/iotdemo/__init__.py
@@ -0,0 +1,10 @@
+"""
+Smart Factory controller APIs
+"""
+
+from .factory_controller import FactoryController
+from .pins import Inputs, Outputs
+from .pyduino import PyDuino
+from .pyft232 import PyFt232
+
+__all__ = ('FactoryController', 'Inputs', 'Outputs', 'PyDuino', 'PyFt232')
diff --git a/iotdemo/debounce.py b/iotdemo/debounce.py
new file mode 100644
index 0000000..a1c158a
--- /dev/null
+++ b/iotdemo/debounce.py
@@ -0,0 +1,164 @@
+"""
+Robust debounce decorator for callbacks like GPIO interrupts.
+
+Features:
+- Leading / trailing edge control
+- Optional max_wait (force-fire) to avoid indefinite deferral
+- Per-key debounce (e.g., per-pin) for shared handlers
+- Thread-safe (uses Lock), time.monotonic() for steady-time
+- Drop-in compatible with previous debounce(wait_sec) usage
+"""
+
+from __future__ import annotations
+import time
+import threading
+from functools import wraps
+from typing import Any, Callable, Optional
+
+
+def debounce(
+ wait_sec: float,
+ *,
+ leading: bool = False,
+ trailing: bool = True,
+ max_wait: Optional[float] = None,
+ key: Optional[Callable[..., Any]] = None,
+ clock: Callable[[], float] = time.monotonic,
+):
+ """
+ Debounce a function so that it’s not called too frequently.
+
+ Args:
+ wait_sec: Quiet period length. Calls within this window are coalesced.
+ leading: If True, call immediately on the first call in a burst.
+ trailing: If True, call after the burst “settles” for wait_sec.
+ max_wait: Force a call if this many seconds have elapsed since the
+ last *invocation* (helps for steady streams).
+ key: Per-key debounce. A function that maps (*args, **kwargs) to a
+ hashable key (e.g., lambda pin, *_: pin). If None, a single
+ global bucket is used.
+ clock: Time source; default is time.monotonic().
+
+ Notes:
+ - If both leading and trailing are True:
+ * first call in a burst fires immediately,
+ * further calls delay a trailing fire until wait_sec after the last call,
+ * but a trailing fire is skipped if there were no calls after the leading fire.
+ - If only leading is True => at most one call per burst, immediate.
+ - If only trailing is True => only the last call args are used.
+ - If max_wait is set, a call is forced at most every max_wait seconds.
+ """
+ if not leading and not trailing:
+ raise ValueError("At least one of leading or trailing must be True.")
+
+ class _Bucket:
+ __slots__ = (
+ "lock", "timer", "last_invoke", "last_call",
+ "pending", "pending_args", "pending_kwargs",
+ )
+ def __init__(self):
+ self.lock = threading.Lock()
+ self.timer: Optional[threading.Timer] = None
+ self.last_invoke: Optional[float] = None # last actual function call time
+ self.last_call: Optional[float] = None # last wrapper call time
+ self.pending: bool = False
+ self.pending_args = ()
+ self.pending_kwargs = {}
+
+ buckets = {}
+ buckets_lock = threading.Lock()
+
+ def _get_bucket(k: Any) -> _Bucket:
+ with buckets_lock:
+ b = buckets.get(k)
+ if b is None:
+ b = _Bucket()
+ buckets[k] = b
+ return b
+
+ def decorator(fn: Callable):
+ @wraps(fn)
+ def wrapper(*args, **kwargs):
+ # derive bucket key
+ k = key(*args, **kwargs) if key else None
+ b = _get_bucket(k)
+ now = clock()
+
+ def _invoke_locked():
+ """Call the function with the latest pending args/kwargs.
+ b.lock must be held by caller."""
+ b.last_invoke = clock()
+ b.pending = False
+ call_args = b.pending_args
+ call_kwargs = b.pending_kwargs
+ # release the lock during the user call to avoid deadlocks
+ b.lock.release()
+ try:
+ return fn(*call_args, **call_kwargs)
+ finally:
+ b.lock.acquire()
+
+ def _schedule_trailing():
+ """(Re)schedule trailing call timer from 'now'."""
+ if b.timer:
+ b.timer.cancel()
+ b.timer = threading.Timer(wait_sec, _trailing_fire)
+ b.timer.daemon = True
+ b.timer.start()
+
+ def _trailing_fire():
+ with b.lock:
+ # fire only if something is pending
+ if not b.pending:
+ return
+ # if max_wait is set and we haven't exceeded it yet AND there
+ # were no new calls, still trailing should run because timer fired
+ _invoke_locked()
+ # after trailing fire, clear timer
+ t = b.timer
+ b.timer = None
+ if t:
+ try:
+ t.cancel()
+ except Exception:
+ pass
+
+ with b.lock:
+ b.last_call = now
+ b.pending = True
+ b.pending_args = args
+ b.pending_kwargs = kwargs
+
+ call_now = False
+
+ # Decide leading fire
+ if leading:
+ if b.last_invoke is None:
+ # first ever
+ call_now = True
+ else:
+ # if enough time has passed since last_invoke -> new burst
+ if (now - b.last_invoke) >= wait_sec and (b.timer is None):
+ call_now = True
+
+ # Decide max_wait forced fire (never conflict with call_now)
+ if (not call_now) and max_wait is not None and b.last_invoke is not None:
+ if (now - b.last_invoke) >= max_wait:
+ call_now = True
+
+ result = None
+ if call_now:
+ result = _invoke_locked()
+ # after a leading/max_wait fire, we may still want trailing if more
+ # calls keep arriving; re-arm the trailing timer if trailing is True
+ if trailing:
+ _schedule_trailing()
+ else:
+ # no immediate call; manage trailing
+ if trailing:
+ _schedule_trailing()
+
+ return result
+
+ return wrapper
+ return decorator
diff --git a/iotdemo/factory_controller.py b/iotdemo/factory_controller.py
new file mode 100644
index 0000000..2708e7b
--- /dev/null
+++ b/iotdemo/factory_controller.py
@@ -0,0 +1,250 @@
+"""
+Smart Factory HW module controller
+"""
+
+from logging import getLogger
+from os import listdir
+from sys import platform
+from time import sleep
+from typing import Optional, Union
+
+from .debounce import debounce
+from .pins import Inputs, Outputs
+from .pyduino import PyDuino
+from .pyft232 import PyFt232
+
+__all__ = ('FactoryController', )
+
+
+class FactoryController:
+ """
+ Factory HW controller class
+ """
+
+ DEV_ON = False
+ DEV_OFF = True
+
+ def __init__(self,
+ port: Optional[Union[str, int]] = None,
+ *,
+ debug: bool = True):
+ self.debug = debug
+ if self.debug:
+ self.logger = getLogger('CONTROLLER')
+
+ self.__force_stop = False
+
+ self.__device = None
+ self.__device_name = None
+
+ # open device
+ self.port = self.__detect_serial(port if port is not None else -1)
+ if self.port is not None and 'ttyUSB' in self.port:
+ try:
+ self.__device = PyFt232(self.port, debug=debug)
+ self.__device_name = 'ft232'
+ except Exception:
+ self.__device = None
+ else:
+ try:
+ self.__device = PyDuino(self.port, debug=debug)
+ self.__device_name = 'arduino'
+ except Exception:
+ self.__device = None
+
+ # Arduino beacon indicator
+ self.red = False
+ self.orange = False
+ self.green = False
+
+ self.led = False
+
+ self.defect_sensor_status = False
+ self.color_sensor_status = False
+
+ # interrupt handler
+ if not self.is_dummy:
+ self.__device.watch(Inputs.START_BUTTON,
+ self.__button_interrupt)
+ self.__device.watch(Inputs.STOP_BUTTON,
+ self.__button_interrupt)
+
+ self.__device.watch(Inputs.PHOTOELECTRIC_SENSOR_1,
+ self.__sensor_interrupt)
+ self.__device.watch(Inputs.PHOTOELECTRIC_SENSOR_2,
+ self.__sensor_interrupt)
+
+ if self.debug:
+ self.logger.info(
+ f'use {"Dummy" if self.is_dummy else "Arduino"} Controller')
+
+ def __detect_serial(self, port):
+ if isinstance(port, str):
+ return port
+
+ if not isinstance(port, int):
+ raise RuntimeError(
+ f'Invalid port argument type: {port} - {type(port)}')
+
+ if platform in {'win32', 'cygwin'}:
+ return f'COM{port}'
+
+ if not platform.startswith('linux'):
+ raise RuntimeError('Not supported OS')
+
+ if port == -1:
+ for path in listdir("/dev"):
+ if path.startswith("ttyACM"):
+ return f"/dev/{path}"
+ else:
+ return f'/dev/ttyACM{port}'
+
+ return None
+
+ def __del__(self):
+ self.close()
+
+ def __enter__(self):
+ if not self.is_dummy:
+ self.system_start()
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ if not self.is_dummy:
+ self.system_stop()
+ self.close()
+
+ def __set(self, pin, value):
+ if self.is_dummy or self.__force_stop:
+ return
+
+ self.__device.set(pin, value)
+
+ def __get(self, pin):
+ if self.is_dummy:
+ return None
+
+ return self.__device.get(pin)
+
+ def __led(self, pin, on):
+ if self.__device_name == 'arduino':
+ self.__set(
+ pin,
+ FactoryController.DEV_ON if on else FactoryController.DEV_OFF)
+
+ def __actuator(self, actuator_id):
+ if self.__device_name == 'arduino':
+ self.__set(actuator_id, FactoryController.DEV_ON)
+ sleep(0.1)
+ self.__set(actuator_id, FactoryController.DEV_OFF)
+ elif self.__device_name == 'ft232':
+ self.__set(PyFt232.PKT_CMD_DETECTION, actuator_id)
+ sleep(0.5)
+ self.__set(PyFt232.PKT_CMD_DETECTION, PyFt232.PKT_CMD_DETECTION_0)
+
+ @debounce(0.1)
+ def __button_interrupt(self, pin: int, value: int, *_):
+ if value != 0 or self.__force_stop:
+ return
+
+ if pin == Inputs.START_BUTTON:
+ self.system_start()
+
+ elif pin == Inputs.STOP_BUTTON:
+ self.system_stop()
+
+ @debounce(0.3)
+ def __sensor_interrupt(self, pin: int, value: int, *_):
+ status = value == 0
+
+ if pin == Inputs.PHOTOELECTRIC_SENSOR_1:
+ self.defect_sensor_status = status
+
+ elif pin == Inputs.PHOTOELECTRIC_SENSOR_2:
+ self.color_sensor_status = status
+
+ ###########################################################################
+ # Public properties
+
+ @property
+ def is_dummy(self):
+ return self.__device is None
+
+ @property
+ def red(self):
+ return bool(self.__get(Outputs.BEACON_RED))
+
+ @red.setter
+ def red(self, on):
+ self.__led(Outputs.BEACON_RED, on)
+
+ @property
+ def orange(self):
+ return bool(self.__get(Outputs.BEACON_ORANGE))
+
+ @orange.setter
+ def orange(self, on):
+ self.__led(Outputs.BEACON_ORANGE, on)
+
+ @property
+ def green(self):
+ return bool(self.__get(Outputs.BEACON_GREEN))
+
+ @green.setter
+ def green(self, on):
+ self.__led(Outputs.BEACON_GREEN, on)
+
+ @property
+ def led(self):
+ return bool(self.__get(Outputs.LED))
+
+ @led.setter
+ def led(self, on):
+ self.__led(Outputs.LED, on)
+
+ @property
+ def conveyor(self):
+ return bool(self.__get(Outputs.CONVEYOR_EN))
+
+ @conveyor.setter
+ def conveyor(self, on):
+ if on:
+ self.__set(Outputs.CONVEYOR_EN, FactoryController.DEV_ON)
+ self.__set(Outputs.CONVEYOR_PWM, 255)
+ else:
+ self.__set(Outputs.CONVEYOR_PWM, 0)
+ self.__set(Outputs.CONVEYOR_EN, FactoryController.DEV_OFF)
+
+ def push_actuator(self, num):
+ if self.__device_name == 'arduino':
+ if num == 1:
+ self.__actuator(Outputs.ACTUATOR_1)
+ else:
+ self.__actuator(Outputs.ACTUATOR_2)
+ elif self.__device_name == 'ft232':
+ self.__actuator(num)
+
+ ###########################################################################
+ # Public methods
+
+ def system_start(self) -> None:
+ if self.debug:
+ self.logger.info("Start System")
+
+ self.conveyor = False
+
+ def system_stop(self) -> None:
+ if self.debug:
+ self.logger.info("Stop System")
+
+ self.conveyor = True
+
+ def close(self):
+ if self.debug:
+ self.logger.info("Shutdown System")
+
+ self.__force_stop = True
+ if self.is_dummy:
+ return
+
+ self.__device.close()
diff --git a/iotdemo/pins.py b/iotdemo/pins.py
new file mode 100644
index 0000000..4da04a1
--- /dev/null
+++ b/iotdemo/pins.py
@@ -0,0 +1,27 @@
+"""
+Input / Output Pin map for the smart factory demo kit
+"""
+from enum import IntEnum
+
+
+class Outputs(IntEnum):
+ BEACON_RED = 2
+ BEACON_ORANGE = 3
+ BEACON_GREEN = 4
+
+ BEACON_BUZZER = 5
+
+ ACTUATOR_1 = 6
+ ACTUATOR_2 = 7
+
+ CONVEYOR_EN = 8
+ CONVEYOR_PWM = 9
+
+ LED = 14
+
+
+class Inputs(IntEnum):
+ START_BUTTON = 10
+ STOP_BUTTON = 11
+ PHOTOELECTRIC_SENSOR_1 = 12
+ PHOTOELECTRIC_SENSOR_2 = 13
diff --git a/iotdemo/pyduino.py b/iotdemo/pyduino.py
new file mode 100644
index 0000000..cc41de8
--- /dev/null
+++ b/iotdemo/pyduino.py
@@ -0,0 +1,294 @@
+"""
+Raw level arduino GPIO controll module
+"""
+
+from logging import getLogger
+from queue import Queue
+from threading import Condition, Lock, Thread
+from typing import Callable, Optional
+
+from serial import Serial
+
+__all__ = ('PyDuino', )
+
+
+def _dummy(*_):
+ pass
+
+
+class PyDuino:
+ """
+ Raw level arduino GPIO controll class
+ """
+
+ PACKET_START = 0x55
+ PACKET_END = 0xaa
+
+ PACKET_ID_INTERRUPT = 0
+ PACKET_ID_MAX = 199
+ PACKET_ID_ERROR = 0xff
+
+ def __init__(self,
+ port: str,
+ baudrate: int = 115200,
+ *,
+ debug: bool = False):
+ self.debug = debug
+ if self.debug:
+ self.logger = getLogger('PYDUINO')
+ self.logger.info("serial port %s (%d)", port, baudrate)
+
+ self.__arduino = None
+ self.__stop_requested = False # to support graceful exit
+ self.__force_stop = False
+ self.__watcher = {}
+ self.__values = [1] * 70 # Pin status
+ for idx in (0, 1):
+ self.__values[idx] = 0xff # NC
+
+ # init arduio
+ arduino = Serial(port=port, baudrate=baudrate, timeout=1)
+ arduino.reset_input_buffer()
+ arduino.reset_output_buffer()
+ self.__arduino = arduino
+
+ self.__init_cv = Condition()
+ self.__init_input_pins = (10, 11, 12, 13)
+ self.__init_count = 0
+
+ self.__rx = Thread(target=self.__receiver)
+ self.__rx.name = 'Arduino Rx'
+ self.__rx.start()
+
+ self.__init()
+
+ self.__tx_lock = Lock()
+ self.__tx_cv = Condition()
+ self.__tx_queue = Queue()
+ self.__tx_packet_id = 1
+ self.__tx = Thread(target=self.__transmitter)
+ self.__tx.name = 'Arduino Tx'
+ self.__tx.start()
+
+ def __del__(self):
+ self.close()
+
+ def __wait_for_input_status(self, pin: int, value: int, _):
+ self.__values[pin] = value
+
+ with self.__init_cv:
+ self.__init_count += 1
+ if self.__init_count == len(self.__init_input_pins):
+ self.__init_cv.notify()
+
+ def __init(self):
+ with self.__init_cv:
+ self.__init_cv.wait()
+
+ for idx in self.__init_input_pins:
+ self.watch(idx)
+
+# pyduino.py - inside class PyDuino
+
+ def __transmitter(self):
+ arduino = self.__arduino
+
+ while True:
+ packet = self.__tx_queue.get()
+ if self.debug:
+ self.logger.info(packet)
+
+ pin, value, packet_id = packet
+ self.__tx_queue.task_done()
+ if pin is None:
+ break # End of Tx
+
+ pin = int(pin) & 0xff
+ value = int(value) & 0xff
+
+ try:
+ arduino.write(bytes([
+ PyDuino.PACKET_START, packet_id, pin, value,
+ PyDuino.PACKET_END
+ ]))
+ except Exception as e:
+ # 치명 오류 시: 강제 정지하고 Tx 종료
+ try:
+ from serial import SerialException
+ except Exception:
+ SerialException = Exception
+ if isinstance(e, (SerialException, OSError)):
+ if self.debug:
+ try: self.logger.warning("Tx write failed / serial down: %s", e)
+ except Exception: pass
+ self.__force_stop = True
+ # set()에서 wait중일 수 있으니 깨워준다
+ try:
+ with self.__tx_cv:
+ self.__tx_cv.notify_all()
+ except Exception:
+ pass
+ break
+ # 기타 오류는 약간 쉬고 재시도(선택)
+ import time; time.sleep(0.02)
+
+ # Rx Thread
+ def __receiver(self):
+ import time
+ # 초기 입력핀 감시 등록(기존 로직 유지)
+ for idx in self.__init_input_pins:
+ self.__values[idx] = 0xff
+ self.watch(idx, self.__wait_for_input_status)
+
+ while self.__force_stop is False:
+ try:
+ packet = self.__receive_packet()
+ except Exception as e:
+ # 시리얼 분리/중복접근 등 치명 오류 → 1회 로깅 후 조용히 종료
+ try:
+ from serial import SerialException
+ except Exception:
+ SerialException = Exception
+ if isinstance(e, (SerialException, OSError)):
+ if self.debug:
+ try: self.logger.warning("Serial disconnected/busy: %s", e)
+ except Exception: pass
+ # 상태 플래그(옵션)
+ try: self._connected = False
+ except Exception: pass
+ break
+ # 기타 오류는 소프트 리트라이
+ time.sleep(0.02)
+ continue
+
+ if packet is None:
+ # 타임아웃/빈 읽기 → 살짝 쉬고 계속
+ time.sleep(0.02)
+ continue
+
+ # 정상 패킷 처리
+ pin, value, packet_id = packet
+ self.__values[pin] = value
+
+ if packet_id == PyDuino.PACKET_ID_INTERRUPT:
+ self.__watcher.get(pin, _dummy)(*packet)
+ else:
+ with self.__tx_cv:
+ self.__tx_cv.notify()
+
+ def __receive_packet(self):
+ arduino = self.__arduino
+
+ # 안전 타임아웃 보정(0/None이면 짧게 설정)
+ try:
+ if getattr(arduino, "timeout", None) in (None, 0):
+ arduino.timeout = 0.2
+ except Exception:
+ pass
+
+ # 첫 바이트 읽기
+ try:
+ data = arduino.read()
+ except Exception:
+ # 상위에서 예외 처리
+ raise
+ if not data or data[0] != PyDuino.PACKET_START:
+ return None
+
+ # 나머지 4바이트 조립(타임아웃이면 None 반환)
+ packet = []
+ while len(packet) != 4:
+ try:
+ data = arduino.read()
+ except Exception:
+ raise
+ if not data:
+ return None # Timeout → 상위 루프가 부드럽게 스킵
+ packet.append(data[0])
+
+ packet_id, pin, value, packet_end = packet
+ if packet_id < 0 or packet_id > PyDuino.PACKET_ID_ERROR:
+ return None
+
+ if pin < 0 or pin > len(self.__values):
+ return None
+
+ if packet_end != PyDuino.PACKET_END:
+ return None
+
+ return (pin, value, packet_id)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
+ def close(self) -> None:
+ """
+ Close existing arduino connection
+ """
+ if self.__arduino is None:
+ return
+
+ self.__stop_requested = True
+ if self.__force_stop:
+ return
+
+ # post end of tx marker
+ self.__tx_queue.put((None, None, None))
+ self.__tx.join()
+
+ # Stop Rx
+ self.__force_stop = True
+ self.__rx.join()
+
+ # close serial
+ self.__arduino.close()
+
+ if self.debug:
+ self.logger.info("stopped")
+
+ # pyduino.py - inside class PyDuino
+
+ def set(self, pin: int, value: int) -> bool:
+ """
+ Set arduino pin with given value
+ """
+ if self.__stop_requested or self.__force_stop:
+ return False
+
+ with self.__tx_lock:
+ packet_id = self.__tx_packet_id
+
+ ok = False
+ with self.__tx_cv:
+ self.__tx_queue.put((pin, value, packet_id))
+ # wait()에 타임아웃 추가 (예: 0.5s)
+ self.__tx_cv.wait(timeout=0.5)
+ ok = not self.__force_stop # Rx/Tx가 죽었으면 False
+
+ self.__tx_packet_id += 1
+ if self.__tx_packet_id == PyDuino.PACKET_ID_MAX:
+ self.__tx_packet_id = 1
+
+ return ok
+
+ def get(self, pin: int) -> int:
+ """
+ Get current arduino pin value from proxy
+ """
+ return self.__values[pin]
+
+ def watch(
+ self,
+ pin: int,
+ callback: Optional[Callable[[int, int, int],
+ None]] = None) -> None:
+ """
+ Set callback function for given pin
+ """
+ if callback is None:
+ callback = _dummy
+
+ self.__watcher[pin] = callback
diff --git a/iotdemo/pyft232.py b/iotdemo/pyft232.py
new file mode 100644
index 0000000..09c53be
--- /dev/null
+++ b/iotdemo/pyft232.py
@@ -0,0 +1,213 @@
+"""
+Raw level FT232 controll module
+"""
+
+from logging import getLogger
+from queue import Queue
+from threading import Condition, Lock, Thread
+from typing import Callable, Optional
+
+from serial import Serial
+
+__all__ = ('PyFt232', )
+
+
+def _dummy(*_):
+ pass
+
+
+class PyFt232:
+ """
+ Raw level FT232 controll class
+ """
+
+ PKT_START = 0xaa
+ PKT_END = 0xaa
+
+ PKT_CMD_ACK = 0x00
+
+ PKT_CMD_START = 0x01
+ PKT_CMD_START_STOP = 0x00
+ PKT_CMD_START_START = 0x01
+
+ PKT_CMD_LOADING = 0x02
+ PKT_CMD_LOADING_STOP = 0x00
+ PKT_CMD_LOADING_LOAD = 0x01
+
+ PKT_CMD_DETECTION = 0x03
+ PKT_CMD_DETECTION_0 = 0x00
+ PKT_CMD_DETECTION_1 = 0x01
+ PKT_CMD_DETECTION_2 = 0x02
+ PKT_CMD_DETECTION_3 = 0x03
+
+ PKT_CMD_SPEED = 0x04
+ PKT_CMD_SPEED_STOP = 0x00
+ PKT_CMD_SPEED_UP = 0x01
+ PKT_CMD_SPEED_DOWN = 0x02
+
+ #PKT_CMD_AUTO_LOADING = 0x05
+ #PKT_CMD_VERSION = 0x64
+
+ PKT_ID_MAX = 199
+ PKT_ID_ERROR = 0xff
+
+ def __init__(self,
+ port: str = '/dev/ttyUSB0',
+ baudrate: int = 19200,
+ *,
+ debug: bool = False):
+ self.debug = debug
+ if self.debug:
+ self.logger = getLogger('PYFT232')
+ self.logger.info("serial port %s (%d)", port, baudrate)
+
+ self.__ft232 = None
+ self.__stop_requested = False # to support graceful exit
+ self.__force_stop = False
+ self.__values = [1] * 14 # Pin status
+ for idx in (0, 1):
+ self.__values[idx] = 0xff # NC
+
+ # init arduio
+ ft232 = Serial(port=port, baudrate=baudrate, timeout=1)
+ ft232.reset_input_buffer()
+ ft232.reset_output_buffer()
+ self.__ft232 = ft232
+
+ self.__rx = Thread(target=self.__receiver)
+ self.__rx.name = 'Arduino Rx'
+ self.__rx.start()
+
+ self.__init()
+
+ self.__tx_lock = Lock()
+ self.__tx_cv = Condition()
+ self.__tx_queue = Queue()
+ self.__tx = Thread(target=self.__transmitter)
+ self.__tx.name = 'Arduino Tx'
+ self.__tx.start()
+
+ def __del__(self):
+ self.close()
+
+ def __checksum(self, length: int, cmd: int, data: list):
+ return ~((length + cmd + sum(data)) & 0xff) & 0xff
+
+ def __init(self):
+ pass
+
+ # Tx Thread
+ def __transmitter(self):
+ ft232 = self.__ft232
+
+ while True:
+ packet = self.__tx_queue.get()
+ if self.debug:
+ self.logger.info(packet)
+
+ if len(packet) != 2:
+ break
+
+ cmd, data = packet
+ self.__tx_queue.task_done()
+ if cmd is None:
+ break # End of Tx
+
+ ft232.write(
+ bytes([
+ PyFt232.PKT_START, 0x05, cmd, data, self.__checksum(0x05, cmd, [data])
+ ]))
+
+ # Rx Thread
+ def __receiver(self):
+ while self.__force_stop is False:
+ packet = self.__receive_packet()
+ if self.debug:
+ self.logger.info(packet)
+ if packet is None:
+ continue
+
+ # TODO : What need to be done?
+ cmd, data = packet
+
+ # run callback
+ with self.__tx_cv:
+ self.__tx_cv.notify()
+
+ def __receive_packet(self):
+ ft232 = self.__ft232
+
+ data = ft232.read()
+ if len(data) == 0 or data[0] != PyFt232.PKT_START:
+ return None
+
+ packet = []
+ while len(packet) != 5:
+ data = ft232.read()
+ if len(data) == 0:
+ return None # Timeout
+
+ packet.append(data[0])
+
+ length, cmd, data, dataerr, checksum = packet
+ if length != 6:
+ return None
+
+ if cmd != 0x00 or dataerr != 0x00:
+ return None
+
+ if self.__checksum(length, cmd, [data, dataerr]) != checksum:
+ return None
+
+ return (cmd, data)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
+
+ def close(self) -> None:
+ """
+ Close existing ft232 connection
+ """
+ if self.__ft232 is None:
+ return
+
+ self.__stop_requested = True
+ if self.__force_stop:
+ return
+
+ # post end of tx marker
+ self.__tx_queue.put((None, None, None))
+ self.__tx.join()
+
+ # Stop Rx
+ self.__force_stop = True
+ self.__rx.join()
+
+ # close serial
+ self.__ft232.close()
+
+ if self.debug:
+ self.logger.info("stopped")
+
+ def set(self, cmd: int, value: int) -> bool:
+ """
+ Set ft232 pin with given value
+ """
+ if self.__stop_requested:
+ return False
+
+ with self.__tx_lock:
+ with self.__tx_cv:
+ self.__tx_queue.put((cmd, value))
+ self.__tx_cv.wait()
+
+ return True
+
+ def get(self, pin: int) -> int:
+ """
+ Get current ft232 pin value from proxy
+ """
+ return self.__values[pin]
diff --git a/logo.png b/logo.png
new file mode 100644
index 0000000..2aca3ac
Binary files /dev/null and b/logo.png differ
diff --git a/requirement.txt b/requirement.txt
new file mode 100644
index 0000000..e5e8c15
--- /dev/null
+++ b/requirement.txt
@@ -0,0 +1,6 @@
+pymysql
+opencv-python
+numpy
+onnxruntime
+tk
+pillow
diff --git a/run.py b/run.py
new file mode 100644
index 0000000..2eced5b
--- /dev/null
+++ b/run.py
@@ -0,0 +1,2452 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os, sys, time, warnings, threading, queue, subprocess
+from datetime import datetime, timedelta
+
+# iotdemo 경로 우선 추가 (사용자 제공 위치)
+try:
+ sys.path.append(os.path.expanduser('./iotdemo'))
+except Exception:
+ pass
+
+# ===== Optional deps (GUI/DB/IO) =====
+import tkinter as tk
+from tkinter import ttk, messagebox
+
+try:
+ import pymysql
+except Exception:
+ pymysql = None
+try:
+ from PIL import Image, ImageTk
+ PIL_OK = True
+except Exception:
+ Image = None; ImageTk = None; PIL_OK = False
+try:
+ import cv2
+except Exception:
+ cv2 = None
+try:
+ import serial
+except Exception:
+ serial = None
+
+# (사용자 환경의 iotdemo.* 가 있을 수도/없을 수도 있으므로 try)
+try:
+ from iotdemo import FactoryController, Inputs, Outputs, PyDuino, PyFt232
+except Exception:
+ FactoryController = Inputs = Outputs = PyDuino = PyFt232 = None
+
+# ====== ONNX/NumPy ======
+import numpy as np
+import onnxruntime as ort
+from math import pi
+
+# ---------- UI colors ----------
+PINK = (203, 80, 165) # bbox
+WHITE = (255,255,255)
+
+# =========================
+# Detection/Seg utilities
+# =========================
+def letterbox(im, size, color=(114,114,114)):
+ h, w = im.shape[:2]
+ r = min(size/h, size/w)
+ nh, nw = int(round(h*r)), int(round(w*r))
+ resized = cv2.resize(im, (nw, nh), interpolation=cv2.INTER_LINEAR)
+ canvas = np.full((size, size, 3), color, dtype=im.dtype)
+ top = (size-nh)//2; left = (size-nw)//2
+ canvas[top:top+nh, left:left+nw] = resized
+ return canvas, r, left, top
+
+def _rounded_rect(img, pt1, pt2, color, thickness=2, r=8):
+ x1,y1 = pt1; x2,y2 = pt2
+ if r<=0:
+ cv2.rectangle(img, pt1, pt2, color, thickness); return
+ cv2.line(img,(x1+r,y1),(x2-r,y1),color,thickness)
+ cv2.line(img,(x1+r,y2),(x2-r,y2),color,thickness)
+ cv2.line(img,(x1,y1+r),(x1,y2-r),color,thickness)
+ cv2.line(img,(x2,y1+r),(x2,y2-r),color,thickness)
+ cv2.ellipse(img,(x1+r,y1+r),(r,r),180,0,90,color,thickness)
+ cv2.ellipse(img,(x2-r,y1+r),(r,r),270,0,90,color,thickness)
+ cv2.ellipse(img,(x1+r,y2-r),(r,r),90 ,0,90,color,thickness)
+ cv2.ellipse(img,(x2-r,y2-r),(r,r),0 ,0,90,color,thickness)
+
+def _label_badge(img, x, y, text, bg, fg=WHITE):
+ font = cv2.FONT_HERSHEY_SIMPLEX; scale=0.55; thick=2; pad=6; r=6
+ (tw,th), _ = cv2.getTextSize(text, font, scale, thick)
+ bx1,by1 = x, max(0, y - th - 2*pad - 6)
+ bx2,by2 = bx1 + tw + 2*pad + 2*r, by1 + th + 2*pad
+ cv2.rectangle(img,(bx1,by1),(bx2,by2),bg,-1)
+ _rounded_rect(img,(bx1,by1),(bx2,by2),bg,2,r)
+ cv2.putText(img, text, (bx1+pad+r//2, by2-pad), font, scale, fg, thick, cv2.LINE_AA)
+
+# [PATCH] 좌측 상단에 멀티라인 텍스트를 그려주는 헬퍼
+def put_lines_top_left(img, lines, colors=None, x=12, y0=22, dy=20):
+ """lines: ['text1','text2',...] colors: [(b,g,r), ...] 또는 None"""
+ font = cv2.FONT_HERSHEY_SIMPLEX; scale=0.55; thick=2
+ for i, text in enumerate(lines):
+ y = y0 + i * dy
+ color = (colors[i] if (colors and i < len(colors)) else (255,255,255))
+ cv2.putText(img, str(text), (x, y), font, scale, color, thick, cv2.LINE_AA)
+
+def draw_box(frame, x1,y1,x2,y2, score, cls, labels, color=PINK):
+ # [PATCH] 박스만 그리고, 텍스트 라벨은 표시하지 않음
+ _rounded_rect(frame, (x1,y1), (x2,y2), color, 2, r=8)
+
+
+def draw_seg_on_roi(roi_bgr, mask_uint8, labels, *, sticker_cid=1, stain_cid=2):
+ h,w = mask_uint8.shape
+ out = roi_bgr.copy()
+ class_plan = [
+ (sticker_cid, {"fill":(0,255,255), "alpha":0.45, "edge":(60,220,60), "badge":(40,180,40), "fallback":"sticker"}),
+ (stain_cid, {"fill":(255,120,0), "alpha":0.45, "edge":(255,210,120), "badge":(200,160,80), "fallback":"stain"}),
+ ]
+ for cid, style in class_plan:
+ if cid < 0:
+ continue
+ m = (mask_uint8 == cid).astype(np.uint8)
+ if m.sum() == 0:
+ continue
+ overlay = out.copy()
+ overlay[m>0] = style["fill"]
+ cv2.addWeighted(overlay, style["alpha"], out, 1-style["alpha"], 0, out)
+ contours, _ = cv2.findContours(m, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+ cv2.drawContours(out, contours, -1, style["edge"], 2)
+ if contours:
+ c = max(contours, key=cv2.contourArea)
+ M = cv2.moments(c)
+ if M["m00"] != 0:
+ cx = int(M["m10"]/M["m00"]); cy = int(M["m01"]/M["m00"])
+ name = labels[cid] if cid < len(labels) else style["fallback"]
+ font=cv2.FONT_HERSHEY_SIMPLEX; scale=0.5; thick=1; pad=4
+ (tw,th), _ = cv2.getTextSize(name, font, scale, thick)
+ bx1,by1 = max(0, cx-tw//2-pad), max(0, cy- th//2 - pad - 12)
+ bx2,by2 = min(w-1, bx1 + tw + 2*pad), min(h-1, by1 + th + 2*pad)
+ cv2.rectangle(out,(bx1,by1),(bx2,by2),style["badge"],-1)
+ cv2.putText(out, name, (bx1+pad, by2-pad), font, scale, WHITE, 1, cv2.LINE_AA)
+ return out
+
+# ---------- Gate & geometry helpers (추가) ----------
+def rect_from_ratio(W, H, wr, hr):
+ bw, bh = int(W*wr), int(H*hr)
+ cx0, cy0 = W//2, H//2
+ return (cx0 - bw//2, cy0 - bh//2, bw, bh)
+
+def rect_from_pct(W, H, L, T, R, B):
+ x1 = int(np.clip(L, 0, 1) * W)
+ y1 = int(np.clip(T, 0, 1) * H)
+ x2 = int(np.clip(R, 0, 1) * W)
+ y2 = int(np.clip(B, 0, 1) * H)
+ if x2 < x1: x1, x2 = x2, x1
+ if y2 < y1: y1, y2 = y2, y1
+ return (x1, y1, x2 - x1, y2 - y1)
+
+def draw_gate_box(frame, rect, color=(0,255,255), thick=2):
+ x, y, w, h = rect
+ cv2.rectangle(frame, (x, y), (x+w, y+h), color, thick)
+
+def circle_fully_inside_rect(cx, cy, r, rect):
+ x, y, w, h = rect
+ x1, y1, x2, y2 = x, y, x+w, y+h
+ return (cx - r >= x1) and (cx + r <= x2) and (cy - r >= y1) and (cy + r <= y2)
+
+# ---------- Detection decoders ----------
+_GRID_CACHE = {}
+def _build_grids(size, strides=(8,16,32)):
+ grids, s_all = [], []
+ for s in strides:
+ ny, nx = size//s, size//s
+ yv, xv = np.meshgrid(np.arange(ny), np.arange(nx), indexing="ij")
+ g = np.stack([xv, yv], -1).reshape(-1,2).astype(np.float32)
+ grids.append(g); s_all.append(np.full((g.shape[0],1), s, np.float32))
+ return np.concatenate(grids,0), np.concatenate(s_all,0)
+
+def _sig(x): return 1/(1+np.exp(-x))
+
+def decode_yolox_raw(pred, size, conf=0.35, num_classes=None):
+ C5,N = pred.shape
+ if num_classes is None: num_classes = C5-5
+ if size not in _GRID_CACHE: _GRID_CACHE[size] = _build_grids(size)
+ grid, strides = _GRID_CACHE[size]
+ if grid.shape[0] != N:
+ M=min(N,grid.shape[0]); pred=pred[:,:M]; N=M
+ px,py,pw,ph = pred[0],pred[1],pred[2],pred[3]
+ pobj=_sig(pred[4]); pcls=_sig(pred[5:5+num_classes])
+ grid=grid[:N]; s=strides[:N,0]
+ cx=(_sig(px)+grid[:,0])*s; cy=(_sig(py)+grid[:,1])*s
+ w=np.exp(pw)*s; h=np.exp(ph)*s
+ x1,y1,x2,y2 = cx-w/2, cy-h/2, cx+w/2, cy+h/2
+ cls_idx=np.argmax(pcls,0); scores=pcls[cls_idx,np.arange(N)]*pobj
+ keep=scores>=conf
+ if not np.any(keep): return np.zeros((0,6),np.float32)
+ return np.stack([x1[keep],y1[keep],x2[keep],y2[keep],scores[keep],cls_idx[keep].astype(np.float32)],1)
+
+def decode_detection_output(arr, img_size, conf=0.35):
+ dets=[]
+ for image_id,label,score,x_min,y_min,x_max,y_max in arr.reshape(-1,7):
+ if score>=conf:
+ dets.append([x_min*img_size,y_min*img_size,x_max*img_size,y_max*img_size,score,label])
+ return np.array(dets,np.float32) if dets else np.zeros((0,6),np.float32)
+
+def decode_box5_plus_cls(out0, out1, size, conf=0.35):
+ B,N,C = out0.shape; assert B==1 and C==5
+ boxes = out0[0].astype(np.float32)
+ clsid = out1[0].astype(np.float32) if (out1.ndim==2 and out1.shape[0]==1) else np.zeros((N,),np.float32)
+ if np.all((boxes[:,:4]>=0)&(boxes[:,:4]<=1.5)): boxes[:,:4]*=size
+ keep = boxes[:,4] >= conf
+ if not np.any(keep): return np.zeros((0,6),np.float32)
+ b=boxes[keep]; c=clsid[keep]
+ return np.concatenate([b[:,:4], b[:,4:5], c[:,None]],1)
+
+def normalize_dets(arr, size):
+ """(N,6) [x1,y1,x2,y2,score,cls] float32로 강제 변환."""
+ if arr is None:
+ return np.zeros((0,6), np.float32)
+ X = np.asarray(arr)
+ if X.ndim == 3 and X.shape[0] == 1:
+ X = X[0]
+ if X.ndim != 2:
+ return np.zeros((0,6), np.float32)
+ if X.shape[1] >= 7: # [image_id,label,score,x1,y1,x2,y2]
+ coords = np.stack([X[:,3], X[:,4], X[:,5], X[:,6]], 1).astype(np.float32)
+ if np.all((coords >= 0.0) & (coords <= 1.5)):
+ coords *= float(size)
+ score = X[:,2:3].astype(np.float32)
+ cls = X[:,1:2].astype(np.float32)
+ Y = np.concatenate([coords, score, cls], 1)
+ return Y.astype(np.float32)
+ if X.shape[1] == 6:
+ Y = X.astype(np.float32)
+ elif X.shape[1] == 5:
+ score = X[:,4:5].astype(np.float32)
+ cls = np.zeros((X.shape[0],1), np.float32)
+ Y = np.concatenate([X[:,:4].astype(np.float32), score, cls], 1)
+ elif X.shape[1] == 4:
+ score = np.ones((X.shape[0],1), np.float32)
+ cls = np.zeros((X.shape[0],1), np.float32)
+ Y = np.concatenate([X[:,:4].astype(np.float32), score, cls], 1)
+ else:
+ Y = X[:,:6].astype(np.float32)
+ if np.all((Y[:,:4] >= 0.0) & (Y[:,:4] <= 1.5)):
+ Y[:,:4] *= float(size)
+ return Y.astype(np.float32)
+
+# ---------- Seg helpers ----------
+def softmax(x, axis=1):
+ x = x - np.max(x, axis=axis, keepdims=True)
+ e = np.exp(x)
+ return e / np.sum(e, axis=axis, keepdims=True)
+
+def _prepare_cls_input(bgr, size, scale=255.0):
+ # 중앙 정사각형 크롭 → resize → RGB → NCHW float32 [0..1]
+ h, w = bgr.shape[:2]
+ s = min(h, w)
+ y0 = (h - s) // 2; x0 = (w - s) // 2
+ crop = bgr[y0:y0+s, x0:x0+s]
+ img = cv2.resize(crop, (size, size), interpolation=cv2.INTER_LINEAR)
+ rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.float32) / float(scale or 255.0)
+ return rgb.transpose(2,0,1)[None]
+
+def _argmax_logits(arr):
+ X = np.asarray(arr)
+ X = X.reshape(1, -1) if X.ndim != 2 else X
+ X = X - X.max(axis=1, keepdims=True)
+ e = np.exp(X); probs = e / e.sum(axis=1, keepdims=True)
+ idx = int(np.argmax(probs[0])); conf = float(probs[0, idx])
+ return idx, conf
+
+# ---------- ORT helpers ----------
+def ort_session(path: str):
+ if not os.path.isfile(path):
+ raise FileNotFoundError(f"Model not found: {path}")
+ providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
+ sess_opt = ort.SessionOptions()
+ sess_opt.intra_op_num_threads = 0
+ sess_opt.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+ return ort.InferenceSession(path, sess_options=sess_opt, providers=providers)
+
+def ort_run(sess, inputs: dict, out_names=None):
+ outs = sess.get_outputs()
+ names = [o.name for o in outs] if out_names is None else out_names
+ return sess.run(names, inputs)
+
+# ---------- seg-map helpers ----------
+def parse_seg_map(s: str, num_classes_hint=3):
+ mapping = {}
+ for part in s.split(","):
+ part = part.strip()
+ if not part:
+ continue
+ a, b = part.split(":")
+ mapping[int(a)] = int(b)
+ for i in range(num_classes_hint):
+ mapping.setdefault(i, i)
+ return mapping
+
+def remap_mask(mask: np.ndarray, m: dict):
+ out = np.zeros_like(mask)
+ for src, dst in m.items():
+ out[mask == src] = dst
+ return out
+
+# ---------- Cap circle + stain ratio ----------
+def try_hough_circle(bgr_roi):
+ gray = cv2.cvtColor(bgr_roi, cv2.COLOR_BGR2GRAY)
+ gray = cv2.medianBlur(gray, 5)
+ circs = cv2.HoughCircles(gray, cv2.HOUGH_GRADIENT, dp=1.2, minDist=40,
+ param1=120, param2=30, minRadius=10, maxRadius=800)
+ if circs is None:
+ return None
+ x, y, r = np.round(circs[0,0]).astype(int)
+ return (x, y, r)
+
+def grade_from_ratio(ratio):
+ if ratio <= 0.0:
+ return (0,255,0)
+ elif ratio <= 30.0:
+ return (0,255,255)
+ else:
+ return (0,0,255)
+
+def draw_cap_circle_and_ratio(frame, cx, cy, diameter_px, stain_mask_full, *, stain_id=2):
+ """
+ 기존 함수 이름 유지.
+ 변경 내용:
+ - 삭제/제거 X (연결성 필터, 열림연산 등 전부 제거)
+ - '픽셀 수'로만 stain 비율 계산
+ - 분모 = 이너서클 픽셀 총계
+ - 분자 = 이너서클 내 stain_id 픽셀 수
+ - rim(림 제외) 옵션만 유지 (기본 6%)
+ """
+ import os
+ H, W = frame.shape[:2]
+ cx = int(np.clip(cx, 0, W - 1))
+ cy = int(np.clip(cy, 0, H - 1))
+
+ # 림 제외 비율
+ rim_frac = float(os.getenv('STAIN_RIM_FRAC', '0.06')) # 6%
+ r_out = int(max(1, round(diameter_px / 2.0)))
+ r_in = max(1, int(round(r_out * (1.0 - np.clip(rim_frac, 0.0, 0.3)))))
+
+ # ---- 분모(캡 전체) 마스크 생성 ----
+ cap_mask = np.zeros((H, W), dtype=np.uint8)
+ cv2.circle(cap_mask, (cx, cy), r_in, 255, -1)
+
+ # ---- 분자(stain 픽셀) 계산 (삭제/제거 무시) ----
+ stain_bin = (stain_mask_full == stain_id).astype(np.uint8) * 255
+ inter = cv2.bitwise_and(stain_bin, cap_mask)
+
+ stain_px = int(cv2.countNonZero(inter))
+ cap_px = int(cv2.countNonZero(cap_mask))
+ ratio = float(stain_px) / max(cap_px, 1) * 100.0
+ ratio = float(np.clip(ratio, 0.0, 100.0))
+
+ # ---- 오버레이 표시 (GUI용) ----
+ color = grade_from_ratio(ratio)
+ cv2.circle(frame, (cx, cy), r_out, (200, 120, 255), 1) # 원 가이드
+ cv2.circle(frame, (cx, cy), r_in, (255, 0, 255), 2) # 분모 영역
+ cv2.line(frame, (max(0, cx - r_in), cy), (min(W - 1, cx + r_in), cy), (255, 0, 255), 2)
+ cv2.putText(frame, f"{ratio:.1f}%", (max(0, cx - r_in), min(H - 10, cy + r_in + 24)),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
+
+ return ratio, stain_px, cap_px
+
+# =========================
+# Detection Worker Thread
+# =========================
+class DetectionWorker(threading.Thread):
+ """
+ 실시간 워커:
+ - 프레임에서 '디텍션만' 수행 (세그는 하지 않음)
+ - 게이트 박스 안 (full/center) 조건 충족 시 캡처
+ - 캡처된 원본 프레임을 snap_q로 전달 (DecideWorker가 Seg/판정)
+ - 프리뷰에는 bbox/게이트만 그려줌 (stain% 미표시)
+ """
+ def __init__(self, get_frame_fn, config, snap_q=None):
+ super().__init__(daemon=True)
+ self.get_frame = get_frame_fn
+ self.cfg = config
+ self._stop = threading.Event()
+ self.last_annot = None
+ self.last_ratio = None # 실시간 ratio는 제공하지 않음
+ self.last_time = 0.0
+
+ # ONNX Runtime (Det only)
+ self.det_sess = ort_session(self.cfg['det_path'])
+ self.det_in_name = self.det_sess.get_inputs()[0].name
+ self.det_out_names = [o.name for o in self.det_sess.get_outputs()]
+
+ # labels
+ self.det_labels = [s.strip() for s in self.cfg['det_labels'].split(",") if s.strip()]
+
+ # gate / capture
+ self._gate_rect = None
+ self._last_capture_ts = 0.0
+ self.snap_q = snap_q if snap_q is not None else queue.Queue(maxsize=16)
+
+ # FPS
+ self._fps_t0 = time.time()
+ self._fps_n = 0
+
+ def stop(self):
+ self._stop.set()
+
+ def run(self):
+ while not self._stop.is_set():
+ frame = self.get_frame()
+ if frame is None:
+ time.sleep(0.01)
+ continue
+ try:
+ out = self._process_frame(frame)
+ self.last_annot = out
+ except Exception:
+ self.last_annot = frame
+ time.sleep(0.005)
+
+ def _process_frame(self, frame):
+ H, W = frame.shape[:2]
+ out_frame = frame.copy()
+
+ # ---------- 게이트 직사각형 초기화 ----------
+ if self._gate_rect is None:
+ roi_pct = self.cfg.get('roi_pct', '')
+ box_str = self.cfg.get('box', '')
+ cap_box = self.cfg.get('capture_box', '0.3,0.3')
+ if roi_pct:
+ L,T,R,B = [float(v) for v in roi_pct.split(",")]
+ self._gate_rect = rect_from_pct(W, H, L, T, R, B)
+ elif box_str:
+ bx,by,bw,bh = [int(v) for v in box_str.split(",")]
+ self._gate_rect = (bx, by, bw, bh)
+ else:
+ wr,hr = [float(x) for x in cap_box.split(",")] if "," in cap_box else (0.3,0.3)
+ self._gate_rect = rect_from_ratio(W, H, wr, hr)
+
+ # ---------- DET ----------
+ d_img, d_r, d_l, d_t = letterbox(frame, self.cfg['det_size'])
+ d_rgb = cv2.cvtColor(d_img, cv2.COLOR_BGR2RGB)
+ d_ten = d_rgb.transpose(2,0,1)[None].astype(np.float32)
+ if self.cfg.get('scale'): d_ten /= self.cfg['scale']
+
+ outs = self.det_sess.run(self.det_out_names, {self.det_in_name: d_ten})
+ d_arrs = []
+ for a in outs:
+ d_arrs.extend(a if isinstance(a, list) else [np.asarray(a)])
+
+ det = None
+ if len(d_arrs) >= 2:
+ a0, a1 = d_arrs[0], d_arrs[1]
+ if a0.ndim==3 and a0.shape[0]==1 and a0.shape[2]==5 and a1.ndim>=2:
+ det = decode_box5_plus_cls(a0, a1, self.cfg['det_size'], self.cfg['det_conf'])
+ elif a1.ndim==3 and a1.shape[0]==1 and a1.shape[2]==5 and a0.ndim>=2:
+ det = decode_box5_plus_cls(a1, a0, self.cfg['det_size'], self.cfg['det_conf'])
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in d_arrs:
+ if (a.ndim==2 and a.shape[-1]>=6): det=a; break
+ if (a.ndim==3 and a.shape[0]==1 and a.shape[-1]>=6): det=a[0]; break
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in d_arrs:
+ if a.ndim==4 and a.shape[-1]==7:
+ det = decode_detection_output(a, self.cfg['det_size'], self.cfg['det_conf'])
+ break
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in d_arrs:
+ if a.ndim==3 and a.shape[0]==1 and (a.shape[1]>=6 or a.shape[2]>=6):
+ raw=a[0]
+ pred = raw if (raw.ndim==2 and raw.shape[0]>=6) else (raw.T if (raw.ndim==2 and raw.shape[1]>=6) else None)
+ if pred is None and raw.ndim==3:
+ tmp=raw.reshape(raw.shape[0],-1); pred=tmp if tmp.shape[0]>=6 else tmp.T
+ if pred is not None:
+ det = decode_yolox_raw(pred, self.cfg['det_size'], self.cfg['det_conf'])
+ break
+ if det is None:
+ det = np.zeros((0,6), np.float32)
+
+ det = normalize_dets(det, self.cfg['det_size'])
+
+ # ---------- 프리뷰: 게이트/박스만 ----------
+ draw_gate_box(out_frame, self._gate_rect, (0,255,255), 2)
+ if det.shape[0] == 0:
+ self._tick_fps(out_frame)
+ return out_frame
+
+ # 최고 score 하나 쓰기
+ x1,y1,x2,y2,score,cls = map(float, det[np.argmax(det[:,4])][:6])
+ gx1=int(np.clip((x1-d_l)/d_r,0,W-1)); gy1=int(np.clip((y1-d_t)/d_r,0,H-1))
+ gx2=int(np.clip((x2-d_l)/d_r,0,W-1)); gy2=int(np.clip((y2-d_t)/d_r,0,H-1))
+ if gx2<=gx1 or gy2<=gy1:
+ self._tick_fps(out_frame)
+ return out_frame
+
+ # 박스/라벨만 표시
+ draw_box(out_frame, gx1, gy1, gx2, gy2, score, cls, self.det_labels, color=PINK)
+
+ # ---------- 게이트 판정 + 캡처 ----------
+ gate_mode = self.cfg.get('gate_mode', 'full')
+ dyn_k = float(self.cfg.get('dynamic_diam_k', 0.90))
+ dyn_diam = max(10.0, dyn_k * min(gx2-gx1, gy2-gy1))
+ r_cap = int(round(dyn_diam / 2.0))
+
+ # 중심점(허프 보정 옵션 없음: Seg가 없으므로 ROI만으로 계산)
+ cx, cy = (gx1+gx2)//2, (gy1+gy2)//2
+ if gate_mode == "full":
+ inside_ok = circle_fully_inside_rect(cx, cy, r_cap, self._gate_rect)
+ else:
+ gx, gy, gw2, gh2 = self._gate_rect
+ inside_ok = (gx <= cx <= gx+gw2 and gy <= cy <= gy+gh2)
+
+ if inside_ok:
+ now = time.time()
+ cooldown = float(self.cfg.get('capture_cooldown', 0.2))
+ if now - self._last_capture_ts >= cooldown:
+ ts_ms = int(now * 1000)
+
+ # 옵션: raw 저장
+ if self.cfg.get('save_raw_caps', 0):
+ os.makedirs("captures/raw", exist_ok=True)
+ cv2.imwrite(f"captures/raw/{ts_ms}.jpg", frame)
+
+ # 스냅샷 전달 (DecideWorker가 Seg/판정)
+ try:
+ self.snap_q.put_nowait((ts_ms, frame.copy()))
+ except queue.Full:
+ try: self.snap_q.get_nowait()
+ except: pass
+ self.snap_q.put_nowait((ts_ms, frame.copy()))
+
+ self._last_capture_ts = now
+
+ self._tick_fps(out_frame)
+ return out_frame
+
+ def _tick_fps(self, out_frame):
+ self._fps_n += 1
+ dt = time.time() - self._fps_t0
+ if dt >= 0.5:
+ self._fps_t0 = time.time()
+ self._fps_n = 0
+
+# =========================
+# Decide Worker (캡쳐 후 오프라인 판정 + 액츄에이터 콜백)
+# =========================
+class DecideWorker(threading.Thread):
+ """
+ 오프라인 워커:
+ - DetectionWorker가 큐에 넣은 스냅샷을 소비
+ - '세그멘테이션 + mask remap + 오염비율 계산 + 등급(OK/NG30/NG80)'
+ - 주석이미지 저장 / CSV 로그 / 액츄에이터 콜백
+ """
+ def __init__(self, snap_q: "queue.Queue", cfg: dict, on_decision_cb):
+ super().__init__(daemon=True)
+ self.snap_q = snap_q
+ self.cfg = cfg
+ self.on_decision = on_decision_cb
+ self._stop = threading.Event()
+
+ # ONNX Runtime (Seg + Det 재사용: 스냅샷에서도 1회 det 수행 → ROI 잡아 Seg)
+ self.det_sess = ort_session(self.cfg['det_path'])
+ self.seg_sess = ort_session(self.cfg['seg_path'])
+ self.det_in = self.det_sess.get_inputs()[0].name
+ self.seg_in = self.seg_sess.get_inputs()[0].name
+ self.det_outs = [o.name for o in self.det_sess.get_outputs()]
+ self.seg_out = self.seg_sess.get_outputs()[0].name
+
+ # label & seg map
+ self.det_labels = [s.strip() for s in self.cfg['det_labels'].split(",") if s.strip()]
+ self.seg_labels = [s.strip() for s in self.cfg['seg_labels'].split(",") if s.strip()]
+ self.seg_map = parse_seg_map(self.cfg['seg_map'], num_classes_hint=max(3, len(self.seg_labels) or 3)) # snap 스타일:contentReference[oaicite:12]{index=12}
+ self.stain_disp_id = self.seg_map.get(self.cfg['stain_id'], self.cfg['stain_id'])
+ self.sticker_disp_id = self.seg_map.get(self.cfg['sticker_id'], self.cfg['sticker_id'])
+
+ # --- [추가] 색상 분류 세션(옵션) ---
+ self.cls_sess = None
+ self.cls_in = None
+ self.cls_out = None
+ self.cls_labels = [s.strip() for s in (self.cfg.get('cls_labels') or '').split(',') if s.strip()]
+ self.cls_size = int(self.cfg.get('cls_size', 224))
+ self.expected_color = (self.cfg.get('expected_color') or '').strip().lower() or None
+ self.cls_map = parse_seg_map(self.cfg.get('cls_map',''), num_classes_hint=max(3, len(self.cls_labels) or 3))
+ try:
+ cls_path = self.cfg.get('cls_path')
+ if cls_path and os.path.isfile(cls_path):
+ self.cls_sess = ort_session(cls_path)
+ self.cls_in = self.cls_sess.get_inputs()[0].name
+ self.cls_out = self.cls_sess.get_outputs()[0].name
+ except Exception:
+ self.cls_sess = None
+
+ def stop(self):
+ self._stop.set()
+
+ def run(self):
+ # 로그/이미지 디렉토리
+ log_dir = "captures/log"
+ os.makedirs(log_dir, exist_ok=True)
+ log_path_v2 = os.path.join(log_dir, "decisions_v2.csv")
+ if self.cfg.get('log_csv', 1):
+ if not os.path.isfile(log_path_v2):
+ with open(log_path_v2, "w") as f:
+ # v2: ts,grade,color,color_conf,sticker,stain_pct
+ f.write("ts,grade,color,color_conf,sticker,stain_pct\n")
+ if self.cfg.get('save_annotated', 1):
+ os.makedirs("captures/annotated", exist_ok=True)
+
+ print("[DecideWorker] Warming up ONNX sessions...")
+ try:
+ # 1. Segmentation Model Warm-up (가장 무거울 가능성 높음)
+ # seg_size에 맞는 더미 입력 생성
+ dummy_seg_input = np.zeros(
+ (1, 3, self.cfg['seg_size'], self.cfg['seg_size']),
+ dtype=np.float32
+ )
+ self.seg_sess.run([self.seg_out], {self.seg_in: dummy_seg_input})
+ print("[DecideWorker] Segmentation model warmed up.")
+
+ # 2. Detection Model Warm-up (DecideWorker에서도 사용)
+ dummy_det_input = np.zeros(
+ (1, 3, self.cfg['det_size'], self.cfg['det_size']),
+ dtype=np.float32
+ )
+ self.det_sess.run(self.det_outs, {self.det_in: dummy_det_input})
+ print("[DecideWorker] Detection model warmed up.")
+
+ # 3. Classification Model Warm-up (옵션, 있다면)
+ if self.cls_sess is not None:
+ dummy_cls_input = np.zeros(
+ (1, 3, self.cls_size, self.cls_size),
+ dtype=np.float32
+ )
+ self.cls_sess.run([self.cls_out], {self.cls_in: dummy_cls_input})
+ print("[DecideWorker] Classification model warmed up.")
+
+ except Exception as e:
+ print(f"[DecideWorker] WARNING: Failed to warm up session: {e}")
+ # ---------------------------------------------------
+
+ while not self._stop.is_set():
+ try:
+ ts_ms, snap_bgr = self.snap_q.get(timeout=0.25)
+ except queue.Empty:
+ continue
+ try:
+ grade, ratio, annotated, has_sticker, pred_color, pred_conf = self._analyze_one(snap_bgr)
+
+ # 저장 파일명에 color 태그 포함
+ if self.cfg.get('save_annotated', 1):
+ color_tag = f"color-{pred_color}" if pred_color else "color-NA"
+ sticker_tag = "yes" if has_sticker else "no"
+ outp = f"captures/annotated/{grade}_{color_tag}_sticker-{sticker_tag}_{ratio:.1f}pct_{ts_ms}.jpg"
+ cv2.imwrite(outp, annotated)
+
+ # CSV(v2): ts,grade,color,color_conf,sticker,stain_pct
+ if self.cfg.get('log_csv', 1):
+ cc = f"{pred_conf:.3f}" if (pred_conf is not None) else ""
+ with open(log_path_v2, "a") as f:
+ f.write(f"{ts_ms},{grade},{pred_color or ''},{cc},{'1' if has_sticker else '0'},{ratio:.3f}\n")
+
+ # 기존 콜백(시그니처 그대로)
+ if callable(self.on_decision):
+ self.on_decision(grade, ratio, has_sticker)
+ except Exception as e:
+ pass
+
+ def _analyze_one(self, img_bgr):
+ """
+ snap_after_seg3.py의 analyze_one_frame 로직을 run.py 스타일로 이식
+ (det → ROI → seg → mask remap → 비율 계산 → 등급)
+ """
+ H, W = img_bgr.shape[:2]
+
+ # ---------- gate (표시용) ----------
+ if self.cfg.get('roi_pct'):
+ L,T,R,B = [float(v) for v in self.cfg['roi_pct'].split(",")]
+ gate = rect_from_pct(W, H, L, T, R, B)
+ elif self.cfg.get('box'):
+ bx,by,bw,bh = [int(v) for v in self.cfg['box'].split(",")]
+ gate = (bx,by,bw,bh)
+ else:
+ wr,hr = [float(x) for x in self.cfg.get('capture_box','0.3,0.3').split(",")]
+ gate = rect_from_ratio(W, H, wr, hr)
+
+ annotated = img_bgr.copy()
+ draw_gate_box(annotated, gate, (0,255,255), 2)
+
+ # ---------- det (스냅샷에서 다시 한 번) ----------
+ d_img, d_r, d_l, d_t = letterbox(img_bgr, self.cfg['det_size'])
+ d_rgb = cv2.cvtColor(d_img, cv2.COLOR_BGR2RGB)
+ d = d_rgb.transpose(2,0,1)[None].astype(np.float32)
+ if self.cfg.get('scale'): d /= self.cfg['scale']
+ outs = self.det_sess.run(self.det_outs, {self.det_in:d})
+ arrs=[]; [arrs.extend(a if isinstance(a,list) else [np.asarray(a)]) for a in outs]
+
+ det=None
+ if len(arrs)>=2:
+ a0,a1=arrs[0],arrs[1]
+ if a0.ndim==3 and a0.shape[0]==1 and a0.shape[2]==5 and a1.ndim>=2:
+ det=decode_box5_plus_cls(a0,a1,self.cfg['det_size'],self.cfg['det_conf'])
+ elif a1.ndim==3 and a1.shape[0]==1 and a1.shape[2]==5 and a0.ndim>=2:
+ det=decode_box5_plus_cls(a1,a0,self.cfg['det_size'],self.cfg['det_conf'])
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in arrs:
+ if (a.ndim==2 and a.shape[-1]>=6): det=a; break
+ if (a.ndim==3 and a.shape[0]==1 and a.shape[-1]>=6): det=a[0]; break
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in arrs:
+ if a.ndim==4 and a.shape[-1]==7:
+ det=decode_detection_output(a,self.cfg['det_size'],self.cfg['det_conf']); break
+ if det is None or (hasattr(det, "size") and det.size==0):
+ for a in arrs:
+ if a.ndim==3 and a.shape[0]==1 and (a.shape[1]>=6 or a.shape[2]>=6):
+ raw=a[0]; pred=raw if (raw.ndim==2 and raw.shape[0]>=6) else (raw.T if (raw.ndim==2 and raw.shape[1]>=6) else None)
+ if pred is None and raw.ndim==3:
+ tmp=raw.reshape(raw.shape[0],-1); pred=tmp if tmp.shape[0]>=6 else tmp.T
+ if pred is not None: det=decode_yolox_raw(pred,self.cfg['det_size'],self.cfg['det_conf']); break
+ if det is None:
+ det=np.zeros((0,6),np.float32)
+
+ if det.shape[0]==0:
+ _label_badge(annotated, 12, 40, "NO_CAP", (40,40,40), WHITE)
+ return "NO_CAP", 0.0, annotated, False,None,None # [PATCH] has_sticker=False
+
+ # 최고 score 1개 ROI
+ x1,y1,x2,y2,score,cls = map(float, det[np.argmax(det[:,4])][:6])
+ gx1=int(np.clip((x1-d_l)/d_r,0,W-1)); gy1=int(np.clip((y1-d_t)/d_r,0,H-1))
+ gx2=int(np.clip((x2-d_l)/d_r,0,W-1)); gy2=int(np.clip((y2-d_t)/d_r,0,H-1))
+ if gx2<=gx1 or gy2<=gy1:
+ return "NO_CAP", 0.0, annotated, False, None, None
+ roi = annotated[gy1:gy2, gx1:gx2]
+ if roi.size==0:
+ return "NO_CAP", 0.0, annotated, False, None, None
+
+ # ---------- [추가] 색상 분류 ----------
+ pred_color = None; pred_conf = None
+ if self.cls_sess is not None:
+ try:
+ cls_in = _prepare_cls_input(roi, self.cls_size, scale=(self.cfg.get('scale') or 255.0))
+ logits = self.cls_sess.run([self.cls_out], {self.cls_in: cls_in})[0]
+ raw_idx, conf = _argmax_logits(logits)
+ idx = self.cls_map.get(int(raw_idx), int(raw_idx)) if hasattr(self, 'cls_map') and isinstance(self.cls_map, dict) else int(raw_idx)
+ pred_color = (self.cls_labels[idx] if 0 <= idx < len(self.cls_labels) else str(idx)).lower()
+ pred_color = (pred_color or 'unknown').lower()
+ pred_conf = conf
+ except Exception:
+ pass
+ # ---------- seg ----------
+ s_img, s_r, s_l, s_t = letterbox(roi, self.cfg['seg_size'])
+ s_rgb = cv2.cvtColor(s_img, cv2.COLOR_BGR2RGB)
+ s = s_rgb.transpose(2,0,1)[None].astype(np.float32)
+ if self.cfg.get('scale'): s /= self.cfg['scale']
+ s_res = self.seg_sess.run([self.seg_out], {self.seg_in:s})[0]
+
+ # multi-class / binary 대응
+ if s_res.ndim==4 and s_res.shape[1]>1:
+ prob = softmax(s_res, axis=1)[0]
+ mask = np.argmax(prob, axis=0).astype(np.uint8)
+ else:
+ mask = (s_res[0,0] > 0.5).astype(np.uint8)
+
+ # 표기가 원하는 class id에 맞게 remap
+ mask = remap_mask(mask, self.seg_map)
+
+ # letterbox 되돌리기 (ROI 크기와 정합)
+ canvas = np.zeros((self.cfg['seg_size'], self.cfg['seg_size']), dtype=np.uint8)
+ nh = int(round(roi.shape[0]*s_r)); nw = int(round(roi.shape[1]*s_r))
+ top = (self.cfg['seg_size']-nh)//2; left=(self.cfg['seg_size']-nw)//2
+ mask_rs = cv2.resize(mask,(nw,nh),interpolation=cv2.INTER_NEAREST)
+ canvas[top:top+nh, left:left+nw] = mask_rs
+ mask_roi = cv2.resize(canvas,(roi.shape[1],roi.shape[0]),interpolation=cv2.INTER_NEAREST)
+
+ # ROI 오버레이
+ annotated[gy1:gy2, gx1:gx2] = draw_seg_on_roi(
+ roi, mask_roi, self.seg_labels,
+ sticker_cid=self.sticker_disp_id,
+ stain_cid=self.stain_disp_id
+ )
+
+ # ---------- [추가] COLOR 배지(프리뷰/저장 공통) ----------
+ color_mismatch = False
+ if self.expected_color and pred_color:
+ color_mismatch = (pred_color != self.expected_color)
+
+ if pred_color:
+ badge_txt = f"COLOR: {pred_color.upper()}"
+ if pred_conf is not None:
+ badge_txt += f" ({pred_conf*100:.0f}%)"
+ _label_badge(
+ annotated,
+ max(0, gx1), max(0, gy1-36),
+ badge_txt,
+ (0,0,255) if color_mismatch else (80,80,80),
+ WHITE
+ )
+ # ---------- STICKER 여부 판단 + 상단 배지 출력 [PATCH] ----------
+ has_sticker = bool(np.any(mask_roi == self.sticker_disp_id))
+ badge_text = f"STICKER: {'YES' if has_sticker else 'NO'}"
+ badge_bg = (40,180,40) if has_sticker else (50,50,200)
+ # 박스 상단 중앙 쯤에 뱃지
+ bx = max(12, gx1 + (gx2 - gx1)//2 - 80)
+ by = max(24, gy1 - 10)
+ _label_badge(annotated, bx, by, badge_text, badge_bg, WHITE)
+
+ # ---------- 오염 비율 (원형 캡 내 비율) ----------
+ # 동적 지름: 디텍션 박스 기반 or 허프 보정
+ dyn_k = float(self.cfg.get('dynamic_diam_k', 0.90))
+ dyn_diam = max(10.0, dyn_k * min(gx2-gx1, gy2-gy1))
+ cx, cy = (gx1+gx2)//2, (gy1+gy2)//2
+ if self.cfg.get('use_hough'):
+ hc = try_hough_circle(annotated[gy1:gy2, gx1:gx2])
+ if hc is not None:
+ rx, ry, rr = hc
+ cx, cy = gx1 + rx, gy1 + ry
+ dyn_diam = max(dyn_diam, rr * 2.0)
+
+ stain_full = np.zeros((H,W), np.uint8)
+ stain_full[gy1:gy2, gx1:gx2] = np.where(mask_roi==self.stain_disp_id, self.stain_disp_id, 0)
+
+ # [PATCH] 픽셀 카운트도 받는다
+ ratio, stain_px, cap_px = draw_cap_circle_and_ratio(
+ annotated, cx, cy, dyn_diam, stain_full, stain_id=self.stain_disp_id
+ )
+ ratio = float(np.clip(ratio, 0.0, 100.0))
+
+ # [PATCH] 좌측 상단에 메트릭 표기
+ sticker_line = f"sticker: {'YES' if has_sticker else 'NO'}"
+ lines = [
+ sticker_line,
+ f"cap_px: {cap_px}",
+ f"stain_px: {stain_px}",
+ f"stain: {ratio:.1f}%",
+ ]
+ colors = [
+ (40,180,40) if has_sticker else (50,50,200),
+ (255,255,255),
+ (255,255,255),
+ (255,255,255),
+ ]
+ # ---------- [추가] 메트릭 첫 줄에 color ----------
+ if pred_color:
+ # 숫자일 경우 라벨 이름으로 변환
+ if isinstance(pred_color, (int, np.integer)) and hasattr(self, "cls_labels"):
+ if 0 <= pred_color < len(self.cls_labels):
+ pred_color = self.cls_labels[pred_color]
+ else:
+ pred_color = str(pred_color)
+ lines.insert(0, f"color: {pred_color} ({pred_conf*100:.0f}% )" if pred_conf is not None else f"color: {pred_color}")
+ colors.insert(0, (0,0,255) if color_mismatch else (255,255,255))
+
+ put_lines_top_left(annotated, lines, colors)
+
+ # ---------- 등급 판정 ----------
+ if color_mismatch:
+ grade = "NG80" # 색상 불일치 = 완전불량(강제)
+ elif not has_sticker:
+ grade = "NG_STICKER" # 스티커 없음 = 불량
+ elif ratio <= 0.0:
+ grade = "OK"
+ elif ratio <= 30.0:
+ grade = "NG30"
+ else:
+ grade = "NG80"
+
+ # 중앙 배지: 등급 | STAIN xx.x%
+ _label_badge(annotated, max(0,cx-60), max(0,cy-80), f"{grade} | STAIN {ratio:.1f}%", (40,40,40), WHITE)
+ draw_box(annotated, gx1, gy1, gx2, gy2, score, cls, self.det_labels, color=PINK)
+
+ return grade, ratio, annotated, has_sticker, pred_color, pred_conf # [PATCH] sticker flag 반환
+
+
+# =========================
+# Hardware / DB / Serial
+# =========================
+class HW:
+ """FactoryController 어댑터: 릴레이 8개 + 센서 4개 + 시스템 시작/정지"""
+ def __init__(self, port=None, debug=True):
+ self.ctrl = None
+ self.is_dummy = True
+ self.dev_name = "dummy"
+ self.state = {i: False for i in range(1,9)}
+ if FactoryController:
+ try:
+ self.ctrl = FactoryController(port=port, debug=debug)
+ self.is_dummy = self.ctrl.is_dummy
+ self.dev_name = getattr(self.ctrl, "_FactoryController__device_name", "unknown")
+ except Exception:
+ self.ctrl = None
+ self.DEV_ON = getattr(FactoryController, "DEV_ON", False)
+ self.DEV_OFF = getattr(FactoryController, "DEV_OFF", True)
+
+ self.relay_map = {
+ 1: getattr(Outputs, "BEACON_RED", None),
+ 2: getattr(Outputs, "BEACON_ORANGE", None),
+ 3: getattr(Outputs, "BEACON_GREEN", None),
+ 4: getattr(Outputs, "BEACON_BUZZER", None),
+ 5: getattr(Outputs, "LED", None),
+ 6: getattr(Outputs, "CONVEYOR_EN", None),
+ 7: getattr(Outputs, "ACTUATOR_1", None),
+ 8: getattr(Outputs, "ACTUATOR_2", None),
+ }
+ self.pwm_pin = getattr(Outputs, "CONVEYOR_PWM", None)
+
+ def set_relay(self, idx:int, on:bool):
+ self.state[idx] = on
+ if not self.ctrl: return
+ pin = self.relay_map.get(idx)
+ if pin is None: return
+ dev = getattr(self.ctrl, "_FactoryController__device", None)
+ if dev is None: return
+
+ # Conveyor EN(active-low), PWM 연동
+ if pin == getattr(Outputs, "CONVEYOR_EN", -1):
+ level = self.DEV_OFF if on else self.DEV_ON
+ try:
+ dev.set(pin, level)
+ if self.pwm_pin is not None:
+ dev.set(self.pwm_pin, 0 if on else 255)
+ except Exception:
+ pass
+ return
+
+ try:
+ dev.set(pin, self.DEV_ON if on else self.DEV_OFF)
+ except Exception:
+ pass
+
+ def system_start(self):
+ if hasattr(self.ctrl, "system_start"):
+ try: self.ctrl.system_start()
+ except Exception: pass
+
+ def system_stop(self):
+ if hasattr(self.ctrl, "system_stop"):
+ try: self.ctrl.system_stop()
+ except Exception: pass
+
+ def get_sensors(self):
+ res = {"start_button": False, "stop_button": False, "sensor_1": False, "sensor_2": False}
+ if not self.ctrl: return res
+ dev = getattr(self.ctrl, "_FactoryController__device", None)
+ name = getattr(self.ctrl, "_FactoryController__device_name", "")
+ if name == "arduino" and dev is not None and Inputs:
+ try:
+ res["start_button"] = (dev.get(Inputs.START_BUTTON) == 0)
+ res["stop_button"] = (dev.get(Inputs.STOP_BUTTON) == 0)
+ res["sensor_1"] = (dev.get(Inputs.PHOTOELECTRIC_SENSOR_1) == 0)
+ res["sensor_2"] = (dev.get(Inputs.PHOTOELECTRIC_SENSOR_2) == 0)
+ except Exception:
+ pass
+ return res
+
+ def get_outputs(self):
+ out = {}
+ if not self.ctrl:
+ return dict(self.state)
+ dev = getattr(self.ctrl, "_FactoryController__device", None)
+ name = getattr(self.ctrl, "_FactoryController__device_name", "")
+ for idx, pin in self.relay_map.items():
+ val = None
+ if name == "arduino" and dev is not None and pin is not None:
+ try:
+ raw = dev.get(pin)
+ if raw in (0,1,True,False):
+ # active-low: 0이면 ON (단, EN은 반대)
+ val = (raw == 0) if pin != getattr(Outputs,"CONVEYOR_EN",-1) else (raw == 1)
+ except Exception:
+ val = None
+ out[idx] = self.state[idx] if val is None else bool(val)
+ return out
+
+ def close(self):
+ try:
+ if self.ctrl: self.ctrl.close()
+ except Exception:
+ pass
+
+# ==== Env / Config ====
+DB_HOST = os.getenv('DB_HOST', '10.10.14.17')
+DB_USER = os.getenv('DB_USER', 'root')
+DB_PASS = os.getenv('DB_PASS', '2453.')
+DB_NAME = os.getenv('DB_NAME', 'iotdb')
+DB_CHARSET = 'utf8mb4'
+
+DB_TABLE_6 = os.getenv('DB_TABLE_6', 'qc_agg6') # ts, id, m1_good,m1_sev,m1_par,m2_good,m2_sev,m2_par
+PARTIAL_WEIGHT = float(os.getenv('PARTIAL_WEIGHT', '0.3'))
+
+# Camera1
+CAM_INDEX = int(os.getenv('CAM_INDEX', '0'))
+CAM_REQ_W = int(os.getenv('CAM_REQ_W', '640'))
+CAM_REQ_H = int(os.getenv('CAM_REQ_H', '480'))
+CAM_VIEW_W = int(os.getenv('CAM_VIEW_W', '560'))
+CAM_VIEW_H = int(os.getenv('CAM_VIEW_H', '315'))
+CAM_FPS = int(os.getenv('CAM_FPS', '30'))
+CAM_BUFFERS = int(os.getenv('CAM_BUFFERS', '1'))
+CAM_FOURCC = os.getenv('CAM_FOURCC', 'MJPG')
+CAM_BACKEND = os.getenv('CAM_BACKEND', '')
+RESAMPLE = os.getenv('RESAMPLE', 'LANCZOS')
+
+# Camera2
+CAM2_INDEX = int(os.getenv('CAM2_INDEX', '2'))
+CAM2_REQ_W = int(os.getenv('CAM2_REQ_W', '640'))
+CAM2_REQ_H = int(os.getenv('CAM2_REQ_H', '480'))
+CAM2_VIEW_W = int(os.getenv('CAM2_VIEW_W', str(CAM_VIEW_W)))
+CAM2_VIEW_H = int(os.getenv('CAM2_VIEW_H', str(CAM_VIEW_H)))
+CAM2_FPS = int(os.getenv('CAM2_FPS', str(CAM_FPS)))
+CAM2_FOURCC = os.getenv('CAM2_FOURCC', 'MJPG')
+CAM2_BACKEND = os.getenv('CAM2_BACKEND', '')
+
+def parse_window(win_str):
+ try:
+ if win_str.endswith('m'): return timedelta(minutes=int(win_str[:-1]))
+ if win_str.endswith('h'): return timedelta(hours=int(win_str[:-1]))
+ if win_str.endswith('d'): return timedelta(days=int(win_str[:-1]))
+ except Exception:
+ pass
+ return timedelta(minutes=5)
+
+# ==== DB helper ====
+def db_query(sql, args=None):
+ if pymysql is None:
+ raise RuntimeError("pymysql not installed. pip install pymysql")
+ conn = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS,
+ db=DB_NAME, charset=DB_CHARSET, autocommit=True)
+ try:
+ with conn.cursor() as c:
+ c.execute(sql, args or ())
+ if sql.strip().lower().startswith("select"):
+ rows = c.fetchall()
+ return rows, [d[0] for d in c.description]
+ else:
+ return (), []
+ finally:
+ conn.close()
+
+# ==== Camera Thread (dual) ====
+class CameraThread(threading.Thread):
+ def __init__(self, index=0, req_w=None, req_h=None, fps=None, fourcc=None, buffers=None, backend=None):
+ super().__init__(daemon=True)
+ self.index=index; self.frame=None; self._stop=threading.Event(); self.cap=None
+ self._lock = threading.Lock()
+ self._req_w = req_w or CAM_REQ_W
+ self._req_h = req_h or CAM_REQ_H
+ self._fps = fps or CAM_FPS
+ self._fourcc = (fourcc or CAM_FOURCC)[:4]
+ self._buffers = buffers if buffers is not None else CAM_BUFFERS
+ self._backend = (backend or CAM_BACKEND).upper()
+
+ def _open(self):
+ if cv2 is None: return None
+ backend_map={'V4L2': cv2.CAP_V4L2, 'DSHOW': cv2.CAP_DSHOW, 'MSMF': cv2.CAP_MSMF}
+ backend_flag = backend_map.get(self._backend, 0)
+
+ cap = cv2.VideoCapture(self.index, backend_flag) if backend_flag else cv2.VideoCapture(self.index)
+ if not (cap and cap.isOpened()):
+ for flag in [cv2.CAP_V4L2, cv2.CAP_MSMF, cv2.CAP_DSHOW]:
+ cap = cv2.VideoCapture(self.index, flag)
+ if cap and cap.isOpened():
+ break
+ if not (cap and cap.isOpened()): return None
+
+ try: cap.set(cv2.CAP_PROP_CONVERT_RGB, 1)
+ except Exception: pass
+ try: cap.set(cv2.CAP_PROP_BUFFERSIZE, self._buffers)
+ except Exception: pass
+ try:
+ fourcc = cv2.VideoWriter_fourcc(*self._fourcc)
+ cap.set(cv2.CAP_PROP_FOURCC, fourcc)
+ except Exception: pass
+ try: cap.set(cv2.CAP_PROP_FRAME_WIDTH, self._req_w)
+ except Exception: pass
+ try: cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self._req_h)
+ except Exception: pass
+ try: cap.set(cv2.CAP_PROP_FPS, self._fps)
+ except Exception: pass
+
+ time.sleep(0.1)
+ ok, f = cap.read()
+ if not ok or f is None:
+ try: cap.release()
+ except Exception: pass
+ return None
+ return cap
+
+ def run(self):
+ while not self._stop.is_set():
+ cap = self._open()
+ self.cap = cap
+ if not cap:
+ time.sleep(1.5); continue
+ interval = 1.0 / max(5, self._fps)
+ while not self._stop.is_set():
+ if not cap.grab():
+ time.sleep(0.05)
+ ok, _ = cap.read()
+ if not ok:
+ try: cap.release()
+ except Exception: pass
+ break
+ ok, frame = cap.retrieve()
+ if ok:
+ with self._lock:
+ self.frame = frame
+ else:
+ try: cap.release()
+ except Exception: pass
+ break
+ time.sleep(interval * 0.5)
+
+ def get_frame(self):
+ with self._lock:
+ return None if self.frame is None else self.frame.copy()
+
+ def stop(self):
+ self._stop.set()
+ try:
+ if self.cap: self.cap.release()
+ except Exception: pass
+
+# ====== 작은 유틸 ======
+def _has_model(cfg: dict) -> bool:
+ return os.path.isfile(cfg.get('det_path','')) and os.path.isfile(cfg.get('seg_path',''))
+
+# =========================
+# App (Tk GUI)
+# =========================
+class App(tk.Tk):
+ def __init__(self):
+ super().__init__()
+ self.title("Factory QC — Dual Camera + DB(6-cols) + Conveyor + Device Panel (overlay + delayed pulse)")
+ self.geometry("1360x980")
+
+ # 신호등(비콘) 상태 추적
+ self._last_beacon = 'none'
+ self._beacon_after = None
+ self._beacon_busy = False # 부팅 시퀀스 중에는 비콘 제어 잠금
+
+ # 하드웨어 컨트롤러
+ self.hw = HW(port=None, debug=True)
+
+ # --- 시스템 실행 상태 / arming ---
+ self._is_running = False # RUN/STOP 램프용
+ self.active_model = 0 # 0=none, 1=M1, 2=M2
+
+ # 컨트롤러 초기화 직후 강제 STOP (시작 깜빡임 방지)
+ try:
+ if self.hw:
+ self.hw.system_stop()
+ self.hw.set_relay(6, False) # Conveyor OFF
+ except Exception:
+ pass
+
+ # 부팅 인디케이터: RED → ORANGE → GREEN → ALL OFF
+ try:
+ self._beacon_boot_chase(step_ms=250, hold_ms=1000)
+ except Exception:
+ pass
+ try:
+ if self.hw:
+ for idx in (1, 2, 3, 4, 5, 7, 8): # 1=R,2=O,3=G,4=Buzzer,5=LED,7/8=Actuators
+ self.hw.set_relay(idx, False)
+ except Exception:
+ pass
+
+ # 시작 후 1초 지연 기간(arming): 자동동작/버튼 RUN 차단
+ self._armed = False
+ self.after(1000, lambda: setattr(self, "_armed", True))
+
+ # Top bar
+ top = ttk.Frame(self); top.pack(fill='x', padx=8, pady=6)
+ try:
+ from PIL import Image, ImageTk
+ logo_path = os.path.join(os.path.dirname(__file__), "logo.png")
+ if os.path.exists(logo_path):
+ logo_img = Image.open(logo_path).resize((42, 30))
+ self._logo_tk = ImageTk.PhotoImage(logo_img)
+ ttk.Label(top, image=self._logo_tk).pack(side="left", padx=(2, 10))
+ else:
+ ttk.Label(top, text="LOGO", font=("Arial", 12, "bold")).pack(side="left", padx=(2, 10))
+ except Exception as e:
+ pass
+
+ self.btn_m1 = ttk.Button(top, text="Model 1 · RUN", command=lambda:self.on_run(1))
+ self.btn_m2 = ttk.Button(top, text="Model 2 · RUN", command=lambda:self.on_run(2))
+ self.btn_stop = ttk.Button(top, text="STOP", command=self.on_stop)
+ self.status_lbl = ttk.Label(top, text="대기")
+ ttk.Label(top, text="RUN:").pack(side='left', padx=(0,6))
+ self.win_var = tk.StringVar(value='5m')
+ self.win_box = ttk.Combobox(top, textvariable=self.win_var, values=['5m','30m','1h','1d'], width=6, state='readonly')
+ self.btn_m1.pack(side='left', padx=4); self.btn_m2.pack(side='left', padx=4); self.btn_stop.pack(side='left', padx=4)
+ self.win_box.pack(side='left'); ttk.Label(top, text=" 상태:").pack(side='left', padx=(16,4)); self.status_lbl.pack(side='left')
+
+ # --- [ADMIN] Hidden toggle hotspot & keybinding ---
+ self._admin_visible = False
+ self._admin_win = None # Toplevel 창 캐시
+ self._hidden_hotspot = tk.Label(top, text=" ", background=self.cget("bg"))
+ self._hidden_hotspot.place(relx=1.0, x=-2, y=2, anchor="ne")
+ self._hidden_hotspot.config(width=1, height=1)
+ self._hidden_hotspot.bind("", lambda e: self._toggle_admin_panel())
+ self.bind_all("", lambda e: self._toggle_admin_panel())
+
+ # 모델별 상태 라벨
+ self.m1_state = tk.StringVar(value='대기')
+ self.m2_state = tk.StringVar(value='대기')
+ ttk.Label(top, text=" | M1:").pack(side='left', padx=(14,2))
+ ttk.Label(top, textvariable=self.m1_state).pack(side='left')
+ ttk.Label(top, text=" M2:").pack(side='left', padx=(8,2))
+ ttk.Label(top, textvariable=self.m2_state).pack(side='left')
+
+ # Grid layout
+ root = ttk.Frame(self); root.pack(fill='both', expand=True, padx=8, pady=6)
+ root.grid_columnconfigure(0, weight=1)
+ root.grid_columnconfigure(1, weight=1)
+ root.grid_rowconfigure(0, weight=1) # 카메라 영역
+ root.grid_rowconfigure(1, weight=3) # 모델/최근(오른쪽) 크게
+ root.grid_rowconfigure(2, weight=1) # 장치 패널
+
+ # KPI 확장용 변수
+ self.k_defw_abs = tk.StringVar(value='-')
+ self.k_tput = tk.StringVar(value='-')
+ self.k_updated = tk.StringVar(value='-')
+
+ # 오버레이에 같이 표기할 실시간 stain% 라벨
+ self.r1_var = tk.StringVar(value='CAM1: - %')
+ self.r2_var = tk.StringVar(value='CAM2: - %')
+
+ # KPI
+ kpi = ttk.LabelFrame(root, text="통합 KPI")
+ kpi.grid(row=0, column=0, sticky='nsew', padx=(0,6), pady=(0,6))
+ self.var_total = tk.StringVar(value='-'); self.var_good = tk.StringVar(value='-')
+ self.var_defw = tk.StringVar(value='-'); self.var_rate = tk.StringVar(value='-')
+ g = ttk.Frame(kpi); g.pack(fill='x', padx=8, pady=8)
+ self._row(g,0,"총 처리", self.var_total); self._row(g,1,"정상품", self.var_good)
+ self._row(g,2,"불량", self.var_defw); self._row(g,3,"불량률", self.var_rate)
+ self._row(g,4,"불량(가중)", self.k_defw_abs)
+ self._row(g,5,"처리속도", self.k_tput)
+ self._row(g,6,"업데이트", self.k_updated)
+ ttk.Label(kpi, textvariable=self.r1_var).pack(anchor="w", padx=10)
+ ttk.Label(kpi, textvariable=self.r2_var).pack(anchor="w", padx=10)
+ self.pb = ttk.Progressbar(kpi, orient='horizontal', length=400, mode='determinate', maximum=100)
+ self.pb.pack(padx=8, pady=(4,10))
+
+ # Cameras (overlay 표시)
+ camf = ttk.LabelFrame(root, text="USB 카메라")
+ camf.grid(row=0, column=1, sticky='nsew', padx=(6,0), pady=(0,6))
+ camgrid = ttk.Frame(camf); camgrid.pack(fill='both', expand=True, padx=6, pady=6)
+ camgrid.grid_columnconfigure(0, weight=1); camgrid.grid_columnconfigure(1, weight=1)
+ self.cam1_lbl = ttk.Label(camgrid, text="CAM1 초기화 중...")
+ self.cam2_lbl = ttk.Label(camgrid, text="CAM2 초기화 중...")
+ self.cam1_lbl.grid(row=0, column=0, padx=4, pady=4, sticky='nsew')
+ self.cam2_lbl.grid(row=0, column=1, padx=4, pady=4, sticky='nsew')
+
+ self.cam1 = CameraThread(index=CAM_INDEX,
+ req_w=CAM_REQ_W, req_h=CAM_REQ_H, fps=CAM_FPS,
+ fourcc=CAM_FOURCC, backend=CAM_BACKEND)
+ self.cam2 = CameraThread(index=CAM2_INDEX,
+ req_w=CAM2_REQ_W, req_h=CAM2_REQ_H, fps=CAM2_FPS,
+ fourcc=CAM2_FOURCC, backend=CAM2_BACKEND)
+ self.cam1.start(); self.cam2.start()
+
+ # Detection workers: CAM1(모델1), CAM2(모델2)
+ self.det_cfg_base = {
+ # 모델 2 입니다.
+ # --- [추가] 색상 분류(classification) 관련 ---
+ 'cls_path': os.path.expanduser(os.getenv('CLASS_MODEL', os.path.join(os.path.dirname(os.path.abspath(__file__)), "class.onnx"))),
+ 'cls_labels': os.getenv('CLASS_LABELS', 'pink,purple'),
+ 'cls_size': int(os.getenv('CLASS_SIZE', '224')),
+ 'cls_map': os.getenv('CLASS_MAP', '1:0,2:1'),
+ 'expected_color': None, # 개별 모델 cfg에서 채움 (model1=pink, model2=purple)
+ 'det_path': os.path.expanduser(os.getenv('DET_MODEL', 'det.onnx')),
+ 'seg_path': os.path.expanduser(os.getenv('SEG_MODEL', 'seg.onnx')),
+ 'det_size': int(os.getenv('DET_SIZE', '640')),
+ 'seg_size': int(os.getenv('SEG_SIZE', '512')),
+ 'det_conf': float(os.getenv('DET_CONF', '0.35')),
+ 'det_labels': os.getenv('DET_LABELS', 'cap'),
+ 'seg_labels': os.getenv('SEG_LABELS', 'background,sticker,stain'),
+ 'seg_map': os.getenv('SEG_MAP', '0:0,1:2,2:1'),
+ 'scale': float(os.getenv('SCALE', '255.0')),
+ 'cap_diam': float(os.getenv('CAP_DIAM', '170')),
+ 'stain_id': int(os.getenv('STAIN_ID', '1')),
+ 'sticker_id': int(os.getenv('STICKER_ID', '2')),
+ 'use_hough': bool(int(os.getenv('USE_HOUGH', '1'))),
+
+ # --- [추가] 게이트/캡쳐 관련 ---
+ 'roi_pct': os.getenv('ROI_PCT', '0.3, 0.3, 0.7, 0.7'), # "L,T,R,B" (0~1)
+ 'box': os.getenv('BOX', ''), # "x,y,w,h" (px)
+ 'capture_box': os.getenv('CAPTURE_BOX', '0.3,0.3'), # 중앙 비율 (백업)
+ 'gate_mode': os.getenv('GATE_MODE', 'full'), # full|center
+ 'capture_cooldown': float(os.getenv('CAPTURE_COOLDOWN', '0.2')),
+ 'dynamic_diam_k': float(os.getenv('DYN_DIAM_K', '0.90')), # 검출박스 기반 지름 스케일
+
+ 'save_raw_caps': int(os.getenv('SAVE_RAW_CAPS', '0')), # 게이트 통과시 원본컷 저장
+ 'save_annotated': int(os.getenv('SAVE_ANNOTATED', '1')), # 판정 결과 이미지 저장
+ 'log_csv': int(os.getenv('LOG_CSV', '1')), # 판정 CSV 로깅
+ }
+
+ model_dir = os.path.dirname(os.path.abspath(__file__))
+ DEFAULT_DET_LOCAL = os.path.join(model_dir, "det.onnx")
+ DEFAULT_SEG_LOCAL = os.path.join(model_dir, "seg.onnx")
+
+ # Model 1 (핑크 전용)
+ self.det_cfg_m1 = dict(self.det_cfg_base)
+ self.det_cfg_m1['det_path'] = os.path.expanduser(os.getenv('DET_MODEL_M1', os.getenv('DET_MODEL', DEFAULT_DET_LOCAL)))
+ self.det_cfg_m1['seg_path'] = os.path.expanduser(os.getenv('SEG_MODEL_M1', os.getenv('SEG_MODEL', DEFAULT_SEG_LOCAL)))
+ self.det_cfg_m1['cls_path'] = os.path.expanduser(os.getenv('CLASS_MODEL_M1', self.det_cfg_base['cls_path']))
+ self.det_cfg_m1['expected_color'] = os.getenv('EXPECTED_COLOR_M1', 'pink')
+
+ # Model 2 (퍼플 전용)
+ self.det_cfg_m2 = dict(self.det_cfg_base)
+ self.det_cfg_m2['det_path'] = os.path.expanduser(os.getenv('DET_MODEL_M2', os.getenv('DET_MODEL', self.det_cfg_base['det_path'])))
+ self.det_cfg_m2['seg_path'] = os.path.expanduser(os.getenv('SEG_MODEL_M2', os.getenv('SEG_MODEL', self.det_cfg_base['seg_path'])))
+ self.det_cfg_m2['cls_path'] = os.path.expanduser(os.getenv('CLASS_MODEL_M2', self.det_cfg_base['cls_path']))
+ self.det_cfg_m2['expected_color'] = os.getenv('EXPECTED_COLOR_M2', 'purple')
+
+
+ self.worker1 = None
+ self.worker2 = None
+
+ self._snap_q = queue.Queue(maxsize=16)
+ self._decider = None
+
+ self._decider = DecideWorker(self._snap_q, self.det_cfg_m1, self._on_decision)
+ self._decider.start()
+
+ self.status_lbl.config(text="대기")
+
+ # Models (요약 패널)
+ model_area = ttk.Frame(root); model_area.grid(row=1, column=0, sticky='nsew', padx=(0,6))
+ model_area.grid_columnconfigure(0, weight=1); model_area.grid_columnconfigure(1, weight=1)
+ self.m1_vars = {k: tk.StringVar(value='-') for k in ['good','sev','par','par_weight','def_weight','overall','rate']}
+ self.m2_vars = {k: tk.StringVar(value='-') for k in ['good','sev','par','par_weight','def_weight','overall','rate']}
+ self._model_panel_simple(model_area, "Model 1", self.m1_vars).grid(row=0, column=0, sticky='nsew', padx=(0,6))
+ self._model_panel_simple(model_area, "Model 2", self.m2_vars).grid(row=0, column=1, sticky='nsew', padx=(6,0))
+
+ # Recent — DB 전용
+ recent = ttk.LabelFrame(root, text="최근 50건")
+ recent.grid(row=1, column=1, rowspan=2, sticky='nsew', padx=(6,0), pady=(0,0))
+ self.tree = ttk.Treeview(
+ recent,
+ columns=("ts","m1_good","m1_sev","m1_par","m2_good","m2_sev","m2_par"),
+ show='headings', height=24
+ )
+ for col, text, w in [
+ ("ts","시간",170),("m1_good","M1 정상",70),("m1_sev","M1 완전 불량",70),("m1_par","M1 부분 불량",70),
+ ("m2_good","M2 정상",70),("m2_sev","M2 완전 불량",70),("m2_par","M2 부분 불량",70)
+ ]:
+ self.tree.heading(col, text=text)
+ self.tree.column(col, width=w, anchor='center')
+ self.tree.pack(fill='both', expand=True, padx=6, pady=6)
+
+ self._fill_recent50_once()
+
+ # (너의 Treeview 생성 블록 바로 아래)
+ self._recent_last_key = None # ← 마지막으로 본 (id,ts) 키 저장
+ self.after(300, self._recent_tick) # ← 0.3초 뒤부터 주기 갱신 시작
+
+ # 장치 패널 — 가로 전체
+ self._build_device_panel(root)
+ # 처음 화면에 STOP 램프가 반드시 들어오도록
+ self._sync_start_stop_lamps()
+
+ # === 액츄에이터 제어용: 디바운스 + 지연(arming) ===
+ self._act1_last = 0.0
+ self._act2_last = 0.0
+ self._act_debounce = float(os.getenv('ACT_DEBOUNCE_SEC', '0.2')) # 0.8초 디바운스
+ self._act_delay = float(os.getenv('ACT_DELAY_SEC', '0.3')) # 0.5초 지연(감지 후)
+
+ # --- [추가] 센서 게이팅용 3초 무장 창 ---
+ self._arm1_deadline = 0.0 # CAM1 유효 마감시각 (epoch)
+ self._arm2_deadline = 0.0 # CAM2 유효 마감시각 (epoch)
+ self._arm1_cond = None # 'sev' 등
+ self._arm2_cond = None # 'par' or 'good'
+
+ self._prev_s1 = False
+ self._prev_s2 = False
+ self._pending1 = None
+ self._pending2 = None
+ self._prev_start_btn = False
+ self._prev_stop_btn = False
+
+ # 기존 짧은 엣지 디바운스(남겨둠, 내부 보조용)
+ self._btn_debounce = float(os.getenv('BTN_DEBOUNCE_SEC', '0.15'))
+ self._last_start_edge = 0.0
+ self._last_stop_edge = 0.0
+
+ # STOP 후 START 잠금(기존 로직)
+ self._stop_latched_until = 0.0
+
+ # ★ 3초 레벨 디바운스(눌린 상태 무시)용 타이머
+ self._start_db_sec = float(os.getenv('START_DEBOUNCE_SEC', '3.0'))
+ self._stop_db_sec = float(os.getenv('STOP_DEBOUNCE_SEC', '3.0'))
+ self._start_ignore_until = 0.0
+ self._stop_ignore_until = 0.0
+
+ # 실시간 카운터(M1,M2) + DB 이벤트 옵션
+ self._counts = {
+ 1: {"good":0, "sev":0, "par":0},
+ 2: {"good":0, "sev":0, "par":0},
+ }
+ self._write_db_on_event = bool(int(os.getenv("WRITE_DB_ON_EVENT", "1")))
+ # timers
+ self.after(120, self.update_camera)
+ self.after(200, self._poll_device_panel)
+ self.after(180, self._logic_loop) # stain% & 센서 연동 로직
+ self.protocol("WM_DELETE_WINDOW", self._on_close)
+
+ # 모델이 없으면 RUN 버튼 비활성화(선택)
+ if not _has_model(self.det_cfg_m1):
+ self.btn_m1.config(state="disabled", text="Model 1 (미설치)")
+ if not _has_model(self.det_cfg_m2):
+ self.btn_m2.config(state="disabled", text="Model 2 (미설치)")
+
+ self._buzz_arm = {"sensor": None, "pattern": None, "deadline": 0.0} # 단일 버저 ARM 슬롯
+
+ def destroy(self):
+ try:
+ if self.worker1: self.worker1.stop()
+ if self.worker2: self.worker2.stop()
+ if hasattr(self, "_decider") and self._decider: self._decider.stop() # [추가]
+ except Exception: pass
+ try: self.cam1.stop(); self.cam2.stop()
+ except Exception: pass
+ try:
+ if hasattr(self, "hw") and self.hw: self.hw.close()
+ except Exception: pass
+ return super().destroy()
+
+ def _row(self, parent, r, name, var):
+ ttk.Label(parent, text=name, width=16).grid(row=r, column=0, sticky='w', padx=4, pady=2)
+ ttk.Label(parent, textvariable=var, width=12).grid(row=r, column=1, sticky='w', padx=4, pady=2)
+
+ def _fill_recent50_once(self):
+ """
+ DB에서 최근 50건을 읽어 Treeview를 '필요할 때만' 갱신한다.
+ - 맨 위 (id, ts)가 이전과 같으면 아무 것도 하지 않음 → 깜빡임 최소화
+ - 바뀌었을 때만 전체 갱신
+ """
+ try:
+ rows, _ = db_query(f'''
+ SELECT id, ts, m1_good, m1_sev, m1_par, m2_good, m2_sev, m2_par
+ FROM {DB_TABLE_6}
+ ORDER BY ts DESC, id DESC
+ LIMIT 50
+ ''')
+ except Exception:
+ rows = []
+
+ # 표시용으로 가공
+ recent_rows = [
+ (
+ ts.strftime('%Y-%m-%d %H:%M:%S'),
+ int(m1g or 0), int(m1s or 0), int(m1p or 0),
+ int(m2g or 0), int(m2s or 0), int(m2p or 0)
+ )
+ for _id, ts, m1g, m1s, m1p, m2g, m2s, m2p in rows
+ ]
+
+ # 변경 확인용 키: 맨 위 레코드의 (id,ts) 조합
+ new_top_key = None
+ if rows:
+ _id0, ts0 = rows[0][0], rows[0][1]
+ new_top_key = f"{_id0}:{ts0.strftime('%Y-%m-%d %H:%M:%S')}"
+
+ # 이전과 동일하면 무변경 → 리턴
+ if new_top_key and getattr(self, "_recent_last_key", None) == new_top_key:
+ return False
+
+ # 여기서만 전체 리프레시 (변경 있을 때만 실행되므로 깜빡임 최소화)
+ try:
+ for iid in self.tree.get_children(""):
+ self.tree.delete(iid)
+ except Exception:
+ pass
+
+ for row in recent_rows:
+ self.tree.insert("", "end", values=row)
+
+ self._recent_last_key = new_top_key
+ return True
+
+ def _recent_tick(self):
+ """최근 50건을 필요할 때만 갱신하고, 다음 주기를 예약한다."""
+ try:
+ self._fill_recent50_once() # 변경 없으면 내부에서 아무 것도 안 함
+ except Exception:
+ pass
+ # 1초 간격으로 폴링 (원하면 1500~2000ms로 늘려도 됨)
+ self.after(1000, self._recent_tick)
+
+ def _model_panel_simple(self, parent, title, vars_dict):
+ lf = ttk.LabelFrame(parent, text=title)
+ g = ttk.Frame(lf); g.pack(fill='x', padx=8, pady=8)
+ items=[("정상품","good"),("완전오염","sev"),("30%오염","par"),
+ ("Partial 가중","par_weight"),("불량(가중)","def_weight"),
+ ("총 처리","overall"),("가중 불량률","rate")]
+ for i,(nm,key) in enumerate(items):
+ ttk.Label(g, text=nm, width=14).grid(row=i, column=0, sticky='w', padx=4, pady=2)
+ ttk.Label(g, textvariable=vars_dict[key], width=10).grid(row=i, column=1, sticky='w', padx=4, pady=2)
+ return lf
+
+ # === RUN/STOP ===
+ def on_run(self, model_id:int):
+ err = None
+ # arming 전에 RUN 금지 (초기 깜빡임/오동작 방지)
+ if not self._armed:
+ self.status_lbl.config(text="초기화 중(잠시 후 RUN 가능)")
+ return
+
+ # 모델 가드
+ if model_id == 1 and not _has_model(self.det_cfg_m1):
+ messagebox.showwarning("Model 1", "Model 1 파일이 없습니다.\n(DET_MODEL_M1 / SEG_MODEL_M1)")
+ return
+ if model_id == 2 and not _has_model(self.det_cfg_m2):
+ messagebox.showwarning("Model 2", "Model 2 파일이 없습니다.\n(DET_MODEL_M2 / SEG_MODEL_M2)")
+ return
+ if not self._restart_workers_for(model_id):
+ return
+
+ self._is_running = True
+ self.active_model = model_id
+ self._sync_start_stop_lamps()
+
+ self._set_beacons(red=False, orange=False, green=False)
+
+ self._pattern_on_model_start(model_id)
+
+ self.status_lbl.config(text=f"Model {model_id} 작동중")
+ if model_id == 1:
+ self.m1_state.set("작동중"); self.m2_state.set("대기")
+ else:
+ self.m1_state.set("대기"); self.m2_state.set("작동중")
+
+ # 백엔드 알림 비동기
+ try:
+ url = os.getenv("BACKEND_URL")
+ if url:
+ def _notify_run(mid):
+ import json, urllib.request
+ req = urllib.request.Request(url.rstrip("/")+"/control",
+ data=json.dumps({"action":"run","model":mid}).encode("utf-8"),
+ headers={"Content-Type":"application/json"})
+ urllib.request.urlopen(req, timeout=1.5)
+ self._bg(_notify_run, model_id)
+ except Exception as e:
+ err = err or e
+
+ # Conveyor ON 확정
+ try:
+ if self.hw: self.hw.set_relay(6, True) # Conveyor ON
+ if hasattr(self, "_relay_vars") and 6 in self._relay_vars:
+ self._relay_vars[6].set(True)
+ self._refresh_relay_visual(6)
+ except Exception:
+ pass
+
+ if err:
+ messagebox.showwarning("제어 알림", f"장치/백엔드 알림 중 오류: {err}")
+
+ def _stop_workers(self):
+ # 두 워커 정지 및 해제
+ for w in (self.worker1, self.worker2):
+ try:
+ if w: w.stop()
+ except Exception:
+ pass
+ for w in (self.worker1, self.worker2):
+ try:
+ if w:w.join(timeout=0.8)
+ except Exception:
+ pass
+ self.worker1 = None
+ self.worker2 = None
+
+ def _start_workers_with_cfg(self, cfg):
+ # 두 카메라 모두 같은 cfg(선택한 모델)로 추론 워커 시작
+ if cv2 is None or not PIL_OK:
+ self.status_lbl.config(text="OpenCV/Pillow 필요");
+ return False
+ try:
+ self.worker1 = DetectionWorker(self.cam1.get_frame, dict(cfg), self._snap_q)
+ self.worker1.start()
+ self.worker2 = DetectionWorker(self.cam2.get_frame, dict(cfg), self._snap_q)
+ self.worker2.start()
+
+ # [추가] 판정 워커 시작 (중복 방지)
+ try:
+ if self._decider:
+ self._decider.stop()
+ self._decider = None
+ except Exception:
+ pass
+ self._decider = DecideWorker(self._snap_q, dict(cfg), self._on_decision)
+ self._decider.start()
+
+ self.status_lbl.config(text="Det/Seg 가동(오버레이 표시) · CAM1/CAM2=같은 모델")
+ return True
+ except Exception as e:
+ self.status_lbl.config(text=f"Worker 시작 실패: {e}")
+ self._stop_workers()
+ return False
+
+ def _restart_workers_for(self, model_id:int):
+ # model_id=1이면 det_cfg_m1, 2면 det_cfg_m2를 두 카메라에 동시에 적용
+ cfg = self.det_cfg_m1 if model_id == 1 else self.det_cfg_m2
+ if not _has_model(cfg):
+ messagebox.showwarning(
+ f"Model {model_id}",
+ f"Model {model_id} 파일이 없습니다.\n(DET_MODEL_M{model_id} / SEG_MODEL_M{model_id})"
+ )
+ return False
+ self._stop_workers()
+ return self._start_workers_with_cfg(cfg)
+
+
+
+ def on_stop(self):
+ err = None
+ self.active_model = 0
+ self._is_running = False
+ self._stop_workers()
+ self._sync_start_stop_lamps()
+ self.status_lbl.config(text="대기")
+ self.m1_state.set("대기"); self.m2_state.set("대기")
+
+ # 백엔드 알림 비동기
+ try:
+ url = os.getenv("BACKEND_URL")
+ if url:
+ def _notify_stop():
+ import json, urllib.request
+ req = urllib.request.Request(url.rstrip("/")+"/control",
+ data=json.dumps({"action":"stop"}).encode("utf-8"),
+ headers={"Content-Type":"application/json"})
+ urllib.request.urlopen(req, timeout=1.5)
+ self._bg(_notify_stop)
+ except Exception as e:
+ err = err or e
+
+ # Conveyor OFF 확정
+ try:
+ if self.hw: self.hw.set_relay(6, False) # Conveyor OFF
+ if hasattr(self, "_relay_vars") and 6 in self._relay_vars:
+ self._relay_vars[6].set(False)
+ self._refresh_relay_visual(6)
+ except Exception:
+ pass
+
+ if err:
+ messagebox.showwarning("제어 알림", f"장치/백엔드 알림 중 오류: {err}")
+
+ # === Camera update (dual, overlay 우선) ===
+ def update_camera(self):
+ try:
+ if cv2 is None or not PIL_OK:
+ self.cam1_lbl.configure(text="(OpenCV/Pillow 필요)")
+ self.cam2_lbl.configure(text="(OpenCV/Pillow 필요)")
+ else:
+ resample = dict(LANCZOS=Image.LANCZOS, BILINEAR=Image.BILINEAR, BICUBIC=Image.BICUBIC).get(RESAMPLE.upper(), Image.LANCZOS)
+
+ # CAM1: worker 오버레이가 있으면 우선 사용
+ f1 = self.worker1.last_annot if (self.worker1 and self.worker1.last_annot is not None) else self.cam1.get_frame()
+ if f1 is not None:
+ rgb1 = cv2.cvtColor(f1, cv2.COLOR_BGR2RGB)
+ img1 = Image.fromarray(rgb1).resize((CAM_VIEW_W, CAM_VIEW_H), resample=resample).copy()
+ self._tk_img1 = ImageTk.PhotoImage(img1)
+ self.cam1_lbl.configure(image=self._tk_img1, text='')
+
+ # CAM2: worker 오버레이 우선
+ f2 = self.worker2.last_annot if (self.worker2 and self.worker2.last_annot is not None) else self.cam2.get_frame()
+ if f2 is not None:
+ rgb2 = cv2.cvtColor(f2, cv2.COLOR_BGR2RGB)
+ img2 = Image.fromarray(rgb2).resize((CAM2_VIEW_W, CAM2_VIEW_H), resample=resample).copy()
+ self._tk_img2 = ImageTk.PhotoImage(img2)
+ self.cam2_lbl.configure(image=self._tk_img2, text='')
+ except Exception as e:
+ self.cam1_lbl.configure(text=f"(CAM1 오류: {e})")
+ self.cam2_lbl.configure(text=f"(CAM2 오류: {e})")
+ finally:
+ self.after(90, self.update_camera)
+
+ # === 백그라운드 실행 헬퍼 ===
+ def _bg(self, target, *args, **kwargs):
+ th = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=True)
+ th.start()
+ return th
+
+ # === 공통 갱신 로직(중복 제거) ===
+ def _apply_simple_node(self, node, vars_dict):
+ sev = int(node['sev']); par = int(node['par']); good = int(node['good'])
+ par_weight = round(par * PARTIAL_WEIGHT, 2)
+ def_weight = round(sev + par_weight, 2)
+ overall = good + sev + par
+ rate = round((def_weight/overall*100.0), 1) if overall>0 else 0.0
+ vars_dict['good'].set(good)
+ vars_dict['sev'].set(sev)
+ vars_dict['par'].set(par)
+ vars_dict['par_weight'].set(f"{par_weight:.2f}")
+ vars_dict['def_weight'].set(f"{def_weight:.2f}")
+ vars_dict['overall'].set(overall)
+ vars_dict['rate'].set(f"{rate}%")
+ return overall, def_weight, good
+
+ def _set_kpi(self, total, good, defw, rate):
+ self.var_total.set(total); self.var_good.set(good)
+ self.var_defw.set(f"{defw:.2f}"); self.var_rate.set(f"{rate}%")
+ self.pb['value'] = min(100, rate)
+
+ def _update_from_counts(self, m1_counts: dict, m2_counts: dict):
+ m1_overall, m1_defw, m1_good = self._apply_simple_node(m1_counts, self.m1_vars)
+ m2_overall, m2_defw, m2_good = self._apply_simple_node(m2_counts, self.m2_vars)
+ total = m1_overall + m2_overall
+ total_good = m1_good + m2_good
+ total_defw = round(m1_defw + m2_defw, 2)
+ rate = round((total_defw/total*100.0), 1) if total>0 else 0.0
+ self._set_kpi(total, total_good, total_defw, rate)
+ self._maybe_blink_beacon(rate)
+ # KPI 확장 항목
+ self.k_defw_abs.set(f"{total_defw:.2f}")
+ win_minutes = max(1e-9, parse_window(self.win_var.get()).total_seconds()/60.0)
+ tput = total / win_minutes if total > 0 else 0.0
+ self.k_tput.set(f"{tput:.1f} 개/분")
+
+ def _win_minutes(self) -> int:
+ s = (self.win_var.get() or '5m').strip().lower()
+ if s.endswith('m'):
+ return max(1, int(s[:-1] or 5))
+ if s.endswith('h'):
+ return max(1, int(s[:-1] or 1) * 60)
+ if s.endswith('d'):
+ return max(1, int(s[:-1] or 1) * 1440)
+ return 5
+
+ def _set_lamp(self, canvas: tk.Canvas, on: bool):
+ if not canvas:
+ return
+ canvas.delete("all")
+ color = "#2ecc71" if on else "#95a5a6"
+ canvas.create_oval(2, 2, 20, 20, fill=color, outline="black")
+
+ def _sync_start_stop_lamps(self):
+ """START/STOP 타일(왼쪽 하단)의 램프를 시스템 상태(_is_running) 기준으로 표시"""
+ try:
+ # START 버튼 타일 = RUN일 때 초록, STOP 버튼 타일 = RUN 아닐 때 초록
+ if hasattr(self, "_sensor_lamps"):
+ self._set_lamp(self._sensor_lamps.get("start_button"), bool(self._is_running))
+ self._set_lamp(self._sensor_lamps.get("stop_button"), not bool(self._is_running))
+ except Exception:
+ pass
+
+ # ==== Beacon helpers ==================================================
+ def _set_beacons(self, red=None, orange=None, green=None):
+ mapping = {1: red, 2: orange, 3: green}
+ for idx, val in mapping.items():
+ if val is None:
+ continue
+ try:
+ self._relay_vars[idx].set(bool(val))
+ if self.hw:
+ self.hw.set_relay(idx, bool(val))
+ self._refresh_relay_visual(idx)
+ except Exception:
+ pass
+
+ def _pulse_beacon(self, kind: str, ms: int = None):
+ """
+ kind: "good" | "par" | "sev"
+ 요청 패턴:
+ - sev(RED): 0.5s ON → 0.5s OFF → 0.5s ON → 0.5s OFF (총 2.0s)
+ - par(ORANGE):0.5s ON → 0.5s OFF (총 1.0s)
+ - good(GREEN):1.0s ON (총 1.0s)
+ """
+ if getattr(self, "_beacon_busy", False):
+ return
+ # 먼저 모두 OFF
+ try:
+ self._set_beacons(red=False, orange=False, green=False)
+ except Exception:
+ pass
+
+ def seq_red():
+ # ON 0.5s
+ self._set_beacons(red=True)
+ self.after(500, lambda: (
+ self._set_beacons(red=False),
+ self.after(500, lambda: (
+ self._set_beacons(red=True),
+ self.after(500, lambda: self._set_beacons(red=False))
+ ))
+ ))
+
+ def seq_orange():
+ self._set_beacons(orange=True)
+ self.after(500, lambda: self._set_beacons(orange=False)) # 0.5s 후 OFF (총 1.0s 간격)
+
+ def seq_green():
+ self._set_beacons(green=True)
+ self.after(1000, lambda: self._set_beacons(green=False))
+
+ if kind == "sev":
+ seq_red()
+ elif kind == "par":
+ seq_orange()
+ elif kind == "good":
+ seq_green()
+ else:
+ return
+
+ # === 모델 시작 패턴: 비프 + GREEN 점멸 ===============================
+ def _beep(self, n=1, dur=0.1, gap=0.1):
+ """릴레이 #4 (Buzzer) 짧게 n회 비프. GUI 블로킹 방지: after 체인."""
+ if not self.hw:
+ return
+ total_steps = n * 2 # ON/OFF 세트
+ def step(i=0):
+ try:
+ on = (i % 2 == 0)
+ self.hw.set_relay(4, on)
+ if 4 in self._relay_vars:
+ self._relay_vars[4].set(on); self._refresh_relay_visual(4)
+ except Exception:
+ pass
+ if i + 1 < total_steps:
+ self.after(int(1000*(dur if on else gap)), lambda: step(i+1))
+ else:
+ # 마지막은 확실히 OFF
+ try:
+ self.hw.set_relay(4, False)
+ if 4 in self._relay_vars:
+ self._relay_vars[4].set(False); self._refresh_relay_visual(4)
+ except Exception:
+ pass
+ step(0)
+
+ def _blink_green(self, count=1, on_ms=500, off_ms=200):
+ """릴레이 #3 (Green Beacon) 점멸 count회."""
+ if not self.hw:
+ return
+ def one_blink(k):
+ # ON
+ try:
+ self._set_beacons(green=True)
+ except Exception:
+ pass
+ # OFF 예약
+ self.after(on_ms, lambda: (
+ self._set_beacons(green=False),
+ self.after(off_ms, lambda: one_blink(k-1) if k-1>0 else None)
+ ))
+ # 먼저 모두 OFF
+ try:
+ self._set_beacons(red=False, orange=False, green=False)
+ except Exception:
+ pass
+ one_blink(count)
+
+ def _pattern_on_model_start(self, model_id:int):
+ """Model 1: 비프 1회 + 초록 0.5s 1회 / Model 2: 비프 2회 + 초록 0.5s 2회"""
+ if model_id == 1:
+ self._beep(n=1, dur=0.1, gap=0.1)
+ self._blink_green(count=1, on_ms=500, off_ms=200)
+ else:
+ self._beep(n=2, dur=0.1, gap=0.1)
+ self._blink_green(count=2, on_ms=500, off_ms=200)
+
+ def _maybe_blink_beacon(self, rate: float, on_ms: int = 800):
+ # 부팅 시퀀스(혹은 다른 배타적 제어) 중엔 개입 금지
+ if getattr(self, "_beacon_busy", False):
+ return
+ desired = 'none'
+ if rate >= 100.0:
+ desired = 'red'
+ elif rate >= 30.0:
+ desired = 'orange'
+ elif rate == 0.0:
+ desired = 'green'
+
+ if desired == self._last_beacon:
+ return
+
+ self._last_beacon = desired
+ if self._beacon_after:
+ try: self.after_cancel(self._beacon_after)
+ except Exception: pass
+ self._beacon_after = None
+
+ self._set_beacons(red=False, orange=False, green=False)
+ if desired == 'none':
+ return
+ if desired == 'red':
+ self._set_beacons(red=True)
+ elif desired == 'orange':
+ self._set_beacons(orange=True)
+ elif desired == 'green':
+ self._set_beacons(green=True)
+ try:
+ self._beacon_after = self.after(on_ms, lambda: self._set_beacons(red=False, orange=False, green=False))
+ except Exception:
+ pass
+
+
+
+ def _beacon_boot_chase(self, step_ms: int = 250, hold_ms: int = 1000):
+ """
+ 앱 시작시: RED -> ORANGE -> GREEN 순서로 "누적" 켠 뒤,
+ GREEN까지 켜진 상태로 hold_ms 유지하고 모두 OFF.
+ """
+ # === 잠금 시작 ===
+ self._beacon_busy = True
+
+ # 기존에 깜빡임 예약이 걸려 있었다면 취소 (중복 개입 차단)
+ try:
+ if getattr(self, "_beacon_after", None):
+ self.after_cancel(self._beacon_after)
+ self._beacon_after = None
+ except Exception:
+ pass
+
+ try:
+ # 모두 끈 상태에서 시작
+ self._set_beacons(red=False, orange=False, green=False)
+ except Exception:
+ pass
+
+ def on_red():
+ # RED 켜고 유지
+ self._set_beacons(red=True, orange=False, green=False)
+ self.after(step_ms, on_orange)
+
+ def on_orange():
+ # ORANGE만 True 전달(RED는 None이라 그대로 유지)
+ self._set_beacons(red=True, orange=True, green=False)
+ self.after(step_ms, on_green)
+
+ def on_green():
+ # GREEN만 True 전달(RED/ORANGE는 None이라 그대로 유지)
+ self._set_beacons(red=True, orange=True, green=True)
+ # 모든 불이 켜진 상태를 hold_ms 유지
+ self.after(hold_ms, all_off)
+
+ def all_off():
+ self._set_beacons(red=False, orange=False, green=False)
+ # === 잠금 해제 ===
+ self._beacon_busy = False
+ # 살짝 딜레이 후 시작
+ self.after(50, on_red)
+
+
+ # ======================================================================
+ # App 클래스 안에 추가 (기존 _set_recent가 있다면 이 버전으로 교체)
+ def _set_recent(self, recent_rows):
+ """
+ recent_rows: 최신이 앞쪽에 오는 리스트.
+ 트리에 데이터가 없을 때만 전체 채우고,
+ 그 외에는 깜빡임 없이 차분 적용(_set_recent_diff)으로 위임.
+ """
+ try:
+ children = self.tree.get_children("")
+ except Exception:
+ children = ()
+ if not children:
+ # 최초 1회만 풀 렌더 (깜빡임 거의 없음)
+ for row in recent_rows:
+ self.tree.insert("", "end", values=row)
+ return
+ # 이후부터는 차분 적용
+ self._set_recent_diff(recent_rows)
+
+ def _update_row_if_changed(self, iid, new_values):
+ """iid 행의 값이 new_values와 다르면 그 행만 갱신(깜빡임 없음)."""
+ try:
+ cur = self.tree.item(iid, "values")
+ # values는 튜플/리스트 문자열들이라 문자열 비교로 충분
+ if tuple(map(str, cur)) != tuple(map(str, new_values)):
+ self.tree.item(iid, values=new_values)
+ except Exception:
+ pass
+
+ def _set_recent_diff(self, recent_rows):
+ """
+ 깜빡임 없이 신규만 위로 추가, 기존 동일 타임스탬프는 값만 갱신.
+ 최신이 앞쪽(0번)으로 온다는 가정.
+ """
+ try:
+ children = list(self.tree.get_children(""))
+ except Exception:
+ children = []
+
+ # 현재 트리 최상단 ts
+ top_ts = None
+ if children:
+ try:
+ top_ts = self.tree.item(children[0], "values")[0]
+ except Exception:
+ top_ts = None
+
+ # 1) 위쪽으로 새로 추가할 개수 계산
+ # recent_rows는 최신이 앞쪽이므로, 기존 top_ts를 만날 때까지가 "신규"
+ new_count = 0
+ for row in recent_rows:
+ ts = row[0]
+ if top_ts is not None and str(ts) == str(top_ts):
+ break
+ new_count += 1
+
+ # 2) 신규를 역순으로 맨 위에 insert (시간 오름차순 유지 위해 역순)
+ if new_count > 0:
+ for row in reversed(recent_rows[:new_count]):
+ self.tree.insert("", 0, values=row)
+
+ # 3) 겹치는 구간(=이미 트리에 있는 범위)은 값이 바뀌었으면 그 행만 갱신
+ # recent_rows[new_count:] 를 트리 상단부터 차례로 대응시켜 비교 갱신
+ start = new_count
+ iids = list(self.tree.get_children(""))
+ for idx, row in enumerate(recent_rows[start:], start=0):
+ if idx >= len(iids):
+ break
+ self._update_row_if_changed(iids[idx], row)
+
+ # 4) 행 수 50개로 유지 (초과분은 아래쪽부터 삭제)
+ MAX_N = 50
+ iids = list(self.tree.get_children(""))
+ if len(iids) > MAX_N:
+ for iid in iids[MAX_N:]:
+ self.tree.delete(iid)
+
+ # === 장치 패널(임베디드) ===
+ def _build_device_panel(self, root):
+ panel = ttk.LabelFrame(root, text=f"장치 패널 — {'device panel' if self.hw.is_dummy else self.hw.dev_name}")
+ panel.grid(row=2, column=0, columnspan=1, sticky='nsew', padx=(0,6), pady=(6,0))
+
+ grid = ttk.Frame(panel); grid.pack(fill='both', expand=True, padx=6, pady=6)
+ for r in range(3):
+ grid.rowconfigure(r, weight=1, uniform="devrow")
+ for c in range(4):
+ grid.columnconfigure(c, weight=1, uniform="devcol")
+
+ self._relay_vars = {}
+ self._relay_lamps = {}
+ names = ["Beacon Red","Beacon Orange","Beacon Green","Buzzer","LED","Conveyor","Actuator 1","Actuator 2"]
+
+ for i in range(8):
+ r = i // 4
+ c = i % 4
+ fr = ttk.Labelframe(grid, text=names[i], padding=8)
+ fr.grid(row=r, column=c, padx=4, pady=4, sticky="nsew")
+ var = tk.BooleanVar(value=False)
+ self._relay_vars[i+1] = var
+ lamp = tk.Canvas(fr, width=22, height=22, highlightthickness=0)
+ lamp.grid(row=0, column=0, padx=4, pady=4)
+ self._relay_lamps[i+1] = lamp
+ self._set_lamp(lamp, False)
+ fr.columnconfigure(0, weight=1)
+
+ self._sensor_lamps = {}
+ sensor_titles = [
+ ("START BUTTON", "start_button"),
+ ("STOP BUTTON", "stop_button"),
+ ("SENSOR 1", "sensor_1"),
+ ("SENSOR 2", "sensor_2"),
+ ]
+ for c, (title, key) in enumerate(sensor_titles):
+ fr = ttk.Labelframe(grid, text=title, padding=8)
+ fr.grid(row=2, column=c, padx=4, pady=4, sticky="nsew")
+ lamp = tk.Canvas(fr, width=22, height=22, highlightthickness=0)
+ lamp.grid(row=0, column=0, padx=4, pady=4)
+ self._sensor_lamps[key] = lamp
+ self._set_lamp(lamp, False)
+ fr.columnconfigure(0, weight=1)
+
+ def _refresh_relay_visual(self, idx:int):
+ on = self._relay_vars[idx].get()
+ lamp = self._relay_lamps.get(idx)
+ self._set_lamp(lamp, on)
+
+ def _poll_device_panel(self):
+ try:
+ st = self.hw.get_sensors() if self.hw else {
+ "start_button": False, "stop_button": False, "sensor_1": False, "sensor_2": False
+ }
+ except Exception:
+ st = {"start_button": False, "stop_button": False, "sensor_1": False, "sensor_2": False}
+
+ try:
+ outs = self.hw.get_outputs() if self.hw else {}
+ except Exception:
+ outs = {}
+
+ # 램프 상태 갱신
+ try:
+ for idx, on in outs.items():
+ var = self._relay_vars.get(idx)
+ if var:
+ if var.get() != bool(on):
+ var.set(bool(on))
+ self._refresh_relay_visual(idx)
+ except Exception:
+ pass
+
+ # === 하드웨어 START/STOP 버튼 처리 (3초 레벨 디바운스 + STOP 우선) ===
+ try:
+ b_start = bool(st.get("start_button", False))
+ b_stop = bool(st.get("stop_button", False))
+ now = time.time()
+
+ # ---- STOP: Rising-edge + 3초 레벨 디바운스 ----
+ if b_stop and not self._prev_stop_btn:
+ if now >= self._stop_ignore_until:
+ self._stop_ignore_until = now + self._stop_db_sec # 3초간 추가 입력 무시
+ self._last_stop_edge = now
+ self._stop_latched_until = now + 1.0 # (기존) 1초 START 락
+ # UI STOP 루틴 호출 (워커/컨베이어 정지 포함)
+ self.on_stop()
+ # 안전하게 비콘/컨베이어 확실히 OFF
+ try:
+ self._set_beacons(red=False, orange=False, green=False)
+ except Exception: pass
+ try:
+ if self.hw:
+ self.hw.set_relay(6, False) # Conveyor OFF
+ if hasattr(self.hw, "system_stop"):
+ self.hw.system_stop()
+ if hasattr(self, "_relay_vars") and 6 in self._relay_vars:
+ self._relay_vars[6].set(False)
+ self._refresh_relay_visual(6)
+ except Exception: pass
+
+ # ---- START: Rising-edge + 3초 레벨 디바운스 + 모델 토글 ----
+ if b_start and not self._prev_start_btn:
+ if now >= self._start_ignore_until:
+ self._start_ignore_until = now + self._start_db_sec # 3초간 추가 입력 무시
+ self._last_start_edge = now
+ # STOP과 동시가 아니고, STOP 락 해제되었고, arming 완료만 확인
+ if self._armed and (not b_stop) and (now >= self._stop_latched_until):
+ if not self._is_running or self.active_model == 0:
+ # 처음 시작 → Model 1
+ self.on_run(1)
+ else:
+ # 토글: 1 ↔ 2
+ next_model = 2 if self.active_model == 1 else 1
+ self.on_run(next_model)
+
+ # 이전 상태 저장 (엣지 검출용)
+ self._prev_start_btn = b_start
+ self._prev_stop_btn = b_stop
+ except Exception:
+ pass
+
+ # 센서 램프
+ try:
+ self._sync_start_stop_lamps() # START/STOP은 시스템 상태 기준
+ if "sensor_1" in self._sensor_lamps:
+ self._set_lamp(self._sensor_lamps["sensor_1"], bool(st.get("sensor_1", False)))
+ if "sensor_2" in self._sensor_lamps:
+ self._set_lamp(self._sensor_lamps["sensor_2"], bool(st.get("sensor_2", False)))
+
+ except Exception:
+ pass
+
+ self.after(200, self._poll_device_panel)
+
+ # === Stain% & 센서 연동 로직 (카메라 역할 고정 + 디바운스/지연 + 엣지 스냅샷) ===
+ def _logic_loop(self):
+ """
+ 센서 라이징 엣지에서 stain% 스냅샷을 저장하고 0.5초 뒤 액츄에이터 펄스.
+ - CAM1: r1 >= 31% AND sensor_1 rising → 0.5s 뒤 릴레이7 (kind='sev')
+ - CAM2: 1% <= r2 <= 30% AND sensor_2 rising → 0.5s 뒤 릴레이8 (kind='par')
+ 0%는 동작 없음. 디바운스 적용.
+ """
+ try:
+ st = self.hw.get_sensors() if self.hw else {"sensor_1":False,"sensor_2":False}
+ s1 = bool(st.get("sensor_1", False))
+ s2 = bool(st.get("sensor_2", False))
+ now = time.time()
+
+ # 실시간 stain% 표시(있으면)
+ r1 = self.worker1.last_ratio if self.worker1 else None
+ r2 = self.worker2.last_ratio if self.worker2 else None
+ if r1 is not None: self.r1_var.set(f"CAM1: {r1:.1f}%")
+ if r2 is not None: self.r2_var.set(f"CAM2: {r2:.1f}%")
+
+ # --- SENSOR 1: CAM1 = 완전오염 전용 ---
+ # --- SENSOR 1: CAM1 (무장창 내에서만 유효) ---
+ if s1 and not self._prev_s1 and self._is_running:
+ now = time.time()
+ if (self._arm1_cond == "sev") and (now <= self._arm1_deadline):
+ if (now - self._act1_last) >= self._act_debounce and self._pending1 is None:
+ self._pending1 = {"when": now, "kind": "sev"}
+ delay_ms = int(self._act_delay * 1000) # 0.3s
+ self.after(delay_ms, self._do_act1)
+ # 트리거 이후 즉시 해제(중복 방지)
+ self._arm1_deadline = 0.0
+ self._arm1_cond = None
+
+ # --- SENSOR 2: CAM2 (무장창 내에서만 유효) ---
+ if s2 and not self._prev_s2 and self._is_running:
+ now = time.time()
+ if (self._arm2_cond in ("par", "good")) and (now <= self._arm2_deadline):
+ # par: Act2 + 주황 패턴 / good: 액츄에이터 없이 green 패턴만
+ if self._arm2_cond == "par":
+ if (now - self._act2_last) >= self._act_debounce and self._pending2 is None:
+ self._pending2 = {"when": now, "kind": "par"}
+ delay_ms = int(self._act_delay * 1000) # 0.3s
+ self.after(delay_ms, self._do_act2)
+ elif self._arm2_cond == "good":
+ # 0.3s 지연 후 GREEN 패턴만 1회
+ def _green_only():
+ # DB 카운터(정상품) + 비콘 그린 패턴
+ self._record_event(kind="good")
+ self.after(int(self._act_delay * 1000), _green_only)
+ # 트리거 이후 즉시 해제(중복 방지)
+ self._arm2_deadline = 0.0
+ self._arm2_cond = None
+
+ # 이전 상태 저장
+ self._prev_s1 = s1
+ self._prev_s2 = s2
+
+ except Exception:
+ pass
+ finally:
+ self.after(60, self._logic_loop) # 주기 60ms
+
+ def _do_act1(self):
+ info = self._pending1; self._pending1 = None
+ self._act1_last = time.time()
+ self._pulse_actuator(7, ms=int(os.getenv('ACT1_PULSE_MS', '250')))
+ self._bg(self._beep_pattern, "sev")
+ if info:
+ self._record_event(kind=info.get("kind","sev"))
+
+ def _do_act2(self):
+ info = self._pending2; self._pending2 = None
+ self._act2_last = time.time()
+ self._pulse_actuator(8, ms=int(os.getenv('ACT2_PULSE_MS', '250')))
+ self._bg(self._beep_pattern, "par")
+ if info:
+ self._record_event(kind=info.get("kind","par"))
+
+ def _pulse_actuator(self, relay_idx:int, ms:int=250):
+ """릴레이 idx를 ms 밀리초 동안 ON 후 자동 OFF"""
+ try:
+ self._relay_vars[relay_idx].set(True)
+ if self.hw: self.hw.set_relay(relay_idx, True)
+ self._refresh_relay_visual(relay_idx)
+ self.after(ms, lambda: self._pulse_off(relay_idx))
+ except Exception:
+ pass
+
+ def _pulse_off(self, relay_idx:int):
+ try:
+ self._relay_vars[relay_idx].set(False)
+ if self.hw: self.hw.set_relay(relay_idx, False)
+ self._refresh_relay_visual(relay_idx)
+ except Exception:
+ pass
+
+ def _record_event(self, model:int=None, kind:str="good"):
+ # model 미지정 → 현재 활성 모델로 귀속
+ if model is None:
+ model = self.active_model
+ if model not in (1,2):
+ return # STOP 상태면 기록하지 않음
+
+ node = self._counts[model]
+ if kind == "sev":
+ node["sev"] += 1
+ elif kind == "par":
+ node["par"] += 1
+ else:
+ node["good"] += 1
+
+ # UI 즉시 반영
+ self._update_from_counts(self._counts[1], self._counts[2])
+ # ★ 이벤트에 맞춰 비콘 펄스
+ try:
+ self._pulse_beacon(kind)
+ except Exception:
+ pass
+
+
+ # 선택적 DB 한 줄 INSERT (WRITE_DB_ON_EVENT=1일 때)
+ if self._write_db_on_event and pymysql is not None:
+ try:
+ cols = ("m1_good","m1_sev","m1_par","m2_good","m2_sev","m2_par")
+ vals = [0,0,0,0,0,0]
+ idx_map = {
+ (1,"good"): 0, (1,"sev"): 1, (1,"par"): 2,
+ (2,"good"): 3, (2,"sev"): 4, (2,"par"): 5,
+ }
+ j = idx_map.get((model,kind), None)
+ if j is not None:
+ vals[j] = 1
+ sql = f"INSERT INTO {DB_TABLE_6} ({','.join(cols)}) VALUES (%s,%s,%s,%s,%s,%s)"
+ self._bg(lambda: db_query(sql, vals))
+ except Exception:
+ pass
+
+ def _toggle_admin_panel(self):
+ if self._admin_win and self._admin_win.winfo_exists():
+ self._admin_win.destroy()
+ self._admin_win = None
+ self._admin_visible = False
+ return
+ self._admin_visible = True
+ self._build_admin_panel()
+
+ def _build_admin_panel(self):
+ win = tk.Toplevel(self)
+ self._admin_win = win
+ win.title("관리자 패널")
+ win.geometry("680x320")
+ try:
+ win.attributes("-topmost", True)
+ except Exception:
+ pass
+
+ def make_toggle_cell(parent, title, relay_idx):
+ cell = ttk.LabelFrame(parent, text=title, padding=8)
+ state = {"on": False}
+ def apply(new_state):
+ state["on"] = bool(new_state)
+ try:
+ if self.hw:
+ self.hw.set_relay(relay_idx, state["on"])
+ except Exception:
+ pass
+ btn.config(text=("ON" if state["on"] else "OFF"))
+
+ if relay_idx == 6:
+ self._is_running = state["on"]
+ self._sync_start_stop_lamps()
+ if relay_idx in self._relay_vars:
+ self._relay_vars[relay_idx].set(state["on"])
+ self._refresh_relay_visual(relay_idx)
+
+ btn = ttk.Button(cell, text="OFF", width=10, command=lambda: apply(not state["on"]))
+ btn.grid(row=0, column=0, sticky="nsew", padx=4, pady=4)
+ try:
+ outs = self.hw.get_outputs() if self.hw else {}
+ cur = bool(outs.get(relay_idx, False))
+ apply(cur)
+ except Exception:
+ pass
+ cell.rowconfigure(0, weight=1)
+ cell.columnconfigure(0, weight=1)
+ return cell
+
+ body = ttk.Frame(win); body.pack(fill="both", expand=True, padx=8, pady=8)
+ rows, cols = 2, 4
+ for r in range(rows):
+ body.rowconfigure(r, weight=1, uniform="admin_row")
+ for c in range(cols):
+ body.columnconfigure(c, weight=1, uniform="admin_col")
+
+ items = [
+ ("Beacon RED", 1),
+ ("Beacon ORANGE", 2),
+ ("Beacon GREEN", 3),
+ ("Buzzer", 4),
+ ("LED", 5),
+ ("Conveyor", 6),
+ ("Actuator 1", 7),
+ ("Actuator 2", 8),
+ ]
+ for i, (title, idx) in enumerate(items):
+ r, c = divmod(i, cols)
+ cell = make_toggle_cell(body, title, idx)
+ cell.grid(row=r, column=c, padx=6, pady=6, sticky="nsew")
+
+ def on_close():
+ self._admin_visible = False
+ win.destroy()
+ self._admin_win = None
+ win.protocol("WM_DELETE_WINDOW", on_close)
+
+ def _on_close(self):
+ self.destroy()
+
+ def _buzzer_set(self, on: bool):
+ # UI 패널 기준 릴레이 #4가 버저
+ try:
+ if getattr(self, "hw", None):
+ self.hw.set_relay(4, bool(on))
+ except Exception:
+ pass
+
+ def _beep_pattern(self, pattern: str):
+ # 릴레이 특성 고려: 짧게 '딸깍' 수준
+ import time
+ def pulse(dt=0.1):
+ self._buzzer_set(True)
+ time.sleep(max(0.01, dt))
+ self._buzzer_set(False)
+
+ if pattern == "sev":
+ # 0.1s → (0.5s 휴지) → 0.1s → (0.5s 휴지)
+ pulse(0.1); time.sleep(0.5); pulse(0.1); time.sleep(0.5)
+ elif pattern == "par":
+ # 0.1s → (0.5s 휴지)
+ pulse(0.1); time.sleep(0.5)
+
+ def _maybe_beep(self, sensor_id: int):
+ now = time.time()
+ arm = self._buzz_arm
+ if arm["sensor"] == sensor_id and now <= arm["deadline"] and arm["pattern"]:
+ # 비동기로 울려서 흐름 안막히게
+ self._bg(self._beep_pattern, arm["pattern"])
+ # 1회 사용 후 소진
+ self._buzz_arm = {"sensor": None, "pattern": None, "deadline": 0.0}
+
+ def _on_decision(self, grade, ratio, has_sticker):
+ print(f"[DECISION] {grade}, stain={ratio:.1f}%, sticker={has_sticker}")
+ now = time.time()
+
+ # NEW: DecideWorker가 준 grade 로 액추에이터 제어 (색상 불일치 포함)
+ if grade in ("NG80", "NG_STICKER"):
+ # 완전불량 → Actuator 1 (RED)
+ self._arm1_deadline = now + 3.0
+ self._arm1_cond = "sev"
+ self._buzz_arm = {"sensor": 1, "pattern": "sev", "deadline": now + 3.0}
+
+ elif grade == "NG30":
+ # 부분불량 → Actuator 2 (ORANGE)
+ self._arm2_deadline = now + 3.0
+ self._arm2_cond = "par"
+ self._buzz_arm = {"sensor": 2, "pattern": "par", "deadline": now + 3.0}
+
+ else: # "OK"
+ # 정상품 → GREEN만 (버저 ARM 없음)
+ self._arm2_deadline = now + 3.0
+ self._arm2_cond = "good"
+ self._buzz_arm = {"sensor": None, "pattern": None, "deadline": 0.0}
+
+
+# === Camera controls (3초 지연) ===
+def delayed_camera_setup():
+ time.sleep(3)
+ cmds = [
+ ["v4l2-ctl", "-d", "/dev/video0", "-c", "auto_exposure=1", "-c", "exposure_dynamic_framerate=0",
+ "-c", "white_balance_automatic=0", "-c", "focus_automatic_continuous=0"],
+ ["v4l2-ctl", "-d", "/dev/video0", "-c", "exposure_time_absolute=60", "-c", "gain=128",
+ "-c", "white_balance_temperature=4500", "-c", "focus_absolute=70"],
+ ["v4l2-ctl", "-d", "/dev/video2", "-c", "auto_exposure=1", "-c", "exposure_dynamic_framerate=0",
+ "-c", "white_balance_automatic=0", "-c", "focus_automatic_continuous=0"],
+ ["v4l2-ctl", "-d", "/dev/video2", "-c", "exposure_time_absolute=60", "-c", "gain=128",
+ "-c", "white_balance_temperature=4500", "-c", "focus_absolute=70"],
+ ]
+ for cmd in cmds:
+ subprocess.run(cmd, check=False)
+
+# =========================
+# Entrypoint
+# =========================
+if __name__ == "__main__":
+ warnings.filterwarnings("ignore", category=RuntimeWarning)
+
+ # GUI 시작
+ app = App()
+
+ # ✅ 3초 후 백그라운드로 카메라 세팅
+ threading.Thread(target=delayed_camera_setup, daemon=True).start()
+
+ app.mainloop()
\ No newline at end of file