feat: 통합 테스트 및 샘플 코드 추가

- test_integration.py: QR 토큰 생성 및 라벨 테스트
- samples/barcode_print.py: Brother QL 프린터 예제
- samples/barcode_reader_gui.py: 바코드 리더 GUI 참고 코드

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-23 16:36:41 +09:00
parent 4581ebb7c5
commit b4de6ff791
8 changed files with 3807 additions and 0 deletions

View File

@@ -0,0 +1,692 @@
"""
허니웰 바코드 리더기 GUI 프로그램 (PyQt5)
COM3 포트에서 바코드를 실시간으로 읽어 화면에 표시
MSSQL DB에서 약품 정보 조회 기능 포함
"""
import sys
import serial
import serial.tools.list_ports
from datetime import datetime
from PyQt5.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QPushButton, QTextEdit, QComboBox, QLabel, QGroupBox, QSpinBox,
QCheckBox, QDialog
)
from PyQt5.QtCore import QThread, pyqtSignal, Qt
from PyQt5.QtGui import QFont, QTextCursor, QPixmap
from sqlalchemy import text
# MSSQL 데이터베이스 연결
sys.path.insert(0, '.')
from dbsetup import DatabaseManager
# 바코드 라벨 출력
from barcode_print import print_barcode_label
def parse_gs1_barcode(barcode):
"""
GS1-128 바코드 파싱
Args:
barcode: 원본 바코드 문자열
Returns:
list: 파싱된 바코드 후보 리스트 (우선순위 순)
"""
candidates = [barcode] # 원본 바코드를 첫 번째 후보로
# GS1-128: 01로 시작하는 경우 (GTIN)
if barcode.startswith('01') and len(barcode) >= 16:
# 01 + 14자리 GTIN
gtin14 = barcode[2:16]
candidates.append(gtin14)
# GTIN-14를 GTIN-13으로 변환 (앞자리가 0인 경우)
if gtin14.startswith('0'):
gtin13 = gtin14[1:]
candidates.append(gtin13)
# GS1-128: 01로 시작하지만 13자리인 경우
elif barcode.startswith('01') and len(barcode) == 15:
gtin13 = barcode[2:15]
candidates.append(gtin13)
return candidates
def search_drug_by_barcode(barcode):
"""
바코드로 약품 정보 조회 (MSSQL PM_DRUG.CD_GOODS)
GS1-128 바코드 자동 파싱 지원
Args:
barcode: 바코드 번호
Returns:
tuple: (약품 정보 dict 또는 None, 파싱 정보 dict)
"""
try:
db_manager = DatabaseManager()
engine = db_manager.get_engine('PM_DRUG')
query = text('''
SELECT TOP 1
BARCODE,
GoodsName,
DrugCode,
SplName,
Price,
Saleprice,
SUNG_CODE,
IsUSE
FROM CD_GOODS
WHERE BARCODE = :barcode
AND (GoodsName NOT LIKE N'%(판매중지)%' AND GoodsName NOT LIKE N'%(판매중단)%')
ORDER BY
CASE WHEN IsUSE = '1' THEN 0 ELSE 1 END, -- 1. 사용중인 제품 우선
CASE WHEN Price > 0 THEN 0 ELSE 1 END, -- 2. 가격 정보 있는 제품 우선
CASE WHEN SplName IS NOT NULL AND SplName != '' THEN 0 ELSE 1 END, -- 3. 제조사 정보 있는 제품 우선
DrugCode DESC -- 4. 약품코드 내림차순
''')
# GS1 바코드 파싱
candidates = parse_gs1_barcode(barcode)
parse_info = {
'original': barcode,
'candidates': candidates,
'matched_barcode': None,
'is_gs1': len(candidates) > 1
}
with engine.connect() as conn:
# 여러 후보 바코드로 순차 검색
for candidate in candidates:
result = conn.execute(query, {"barcode": candidate})
row = result.fetchone()
if row:
parse_info['matched_barcode'] = candidate
drug_info = {
'barcode': row.BARCODE,
'goods_name': row.GoodsName,
'drug_code': row.DrugCode,
'manufacturer': row.SplName,
'price': float(row.Price) if row.Price else 0,
'sale_price': float(row.Saleprice) if row.Saleprice else 0,
'sung_code': row.SUNG_CODE if row.SUNG_CODE else ''
}
return drug_info, parse_info
return None, parse_info
except Exception as e:
print(f'[오류] 약품 조회 실패: {e}')
return None, {'original': barcode, 'error': str(e)}
class DrugSearchThread(QThread):
"""약품 정보 조회 전용 백그라운드 스레드"""
# 시그널: (바코드, 타임스탬프, 원본 데이터, 약품 정보, 파싱 정보)
search_complete = pyqtSignal(str, str, bytes, object, object)
def __init__(self, barcode, timestamp, raw_data):
super().__init__()
self.barcode = barcode
self.timestamp = timestamp
self.raw_data = raw_data
def run(self):
"""백그라운드에서 DB 조회"""
drug_info, parse_info = search_drug_by_barcode(self.barcode)
self.search_complete.emit(self.barcode, self.timestamp, self.raw_data, drug_info, parse_info)
class LabelGeneratorThread(QThread):
"""라벨 이미지 생성 전용 백그라운드 스레드"""
# 시그널: (성공 여부, 이미지 경로, 약품명, 에러 메시지)
image_ready = pyqtSignal(bool, str, str, str)
def __init__(self, goods_name, sale_price, preview_mode=False):
super().__init__()
self.goods_name = goods_name
self.sale_price = sale_price
self.preview_mode = preview_mode
def run(self):
"""백그라운드에서 이미지 생성"""
try:
if self.preview_mode:
# 미리보기 모드
success, image_path = print_barcode_label(
self.goods_name,
self.sale_price,
preview_mode=True
)
if success:
self.image_ready.emit(True, image_path, self.goods_name, "")
else:
self.image_ready.emit(False, "", self.goods_name, "이미지 생성 실패")
else:
# 실제 인쇄 모드
success = print_barcode_label(
self.goods_name,
self.sale_price,
preview_mode=False
)
if success:
self.image_ready.emit(True, "", self.goods_name, "")
else:
self.image_ready.emit(False, "", self.goods_name, "라벨 출력 실패")
except Exception as e:
self.image_ready.emit(False, "", self.goods_name, str(e))
class LabelPreviewDialog(QDialog):
"""라벨 미리보기 팝업 창"""
def __init__(self, image_path, goods_name, parent=None):
"""
Args:
image_path: 미리보기 이미지 파일 경로
goods_name: 약품명
parent: 부모 위젯
"""
super().__init__(parent)
self.image_path = image_path
self.goods_name = goods_name
self.init_ui()
def init_ui(self):
"""UI 초기화"""
self.setWindowTitle(f'라벨 미리보기 - {self.goods_name}')
self.setModal(False) # 모달 아님 (계속 스캔 가능)
# 레이아웃
layout = QVBoxLayout()
# 상단 안내 라벨
info_label = QLabel('[미리보기] 실제 인쇄하려면 "미리보기 모드" 체크를 해제하세요.')
info_label.setStyleSheet('color: #2196F3; font-size: 12px; padding: 10px;')
layout.addWidget(info_label)
# 이미지 표시 (QLabel + QPixmap)
pixmap = QPixmap(self.image_path)
# 화면 크기에 맞게 스케일링 (최대 1000px 폭)
if pixmap.width() > 1000:
pixmap = pixmap.scaledToWidth(1000, Qt.SmoothTransformation)
image_label = QLabel()
image_label.setPixmap(pixmap)
image_label.setAlignment(Qt.AlignCenter)
layout.addWidget(image_label)
# 버튼 레이아웃
button_layout = QHBoxLayout()
# 닫기 버튼
close_btn = QPushButton('닫기')
close_btn.setStyleSheet('background-color: #2196F3; color: white; font-weight: bold; padding: 8px 20px;')
close_btn.clicked.connect(self.close)
button_layout.addWidget(close_btn)
layout.addLayout(button_layout)
self.setLayout(layout)
# 창 크기 자동 조정
self.adjustSize()
class BarcodeReaderThread(QThread):
"""바코드 읽기 스레드 (DB 조회 없이 바코드만 읽음)"""
barcode_received = pyqtSignal(str, str, bytes) # 바코드, 시간, 원본 (DB 조회 제외!)
connection_status = pyqtSignal(bool, str) # 연결 상태, 메시지
raw_data_received = pyqtSignal(str) # 시리얼 포트 RAW 데이터 (디버깅용)
def __init__(self, port='COM3', baudrate=115200):
super().__init__()
self.port = port
self.baudrate = baudrate
self.running = False
self.serial_connection = None
def run(self):
"""스레드 실행"""
self.running = True
try:
# 시리얼 포트 열기
self.serial_connection = serial.Serial(
port=self.port,
baudrate=self.baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1
)
self.connection_status.emit(True, f'{self.port} 연결 성공! (속도: {self.baudrate} bps)')
# 바코드 읽기 루프
while self.running:
if self.serial_connection.in_waiting > 0:
buffer_size = self.serial_connection.in_waiting
timestamp_ms = datetime.now().strftime('%H:%M:%S.%f')[:-3]
# 즉시 GUI에 표시
self.raw_data_received.emit(f'[{timestamp_ms}] 버퍼: {buffer_size} bytes')
# 버퍼의 모든 데이터를 한 번에 읽기 (연속 스캔 대응)
all_data = self.serial_connection.read(buffer_size)
# 즉시 GUI에 표시
self.raw_data_received.emit(f' → 읽음: {all_data.hex()} ({len(all_data)} bytes)')
# 디코딩
try:
all_text = all_data.decode('utf-8')
except UnicodeDecodeError:
all_text = all_data.decode('ascii', errors='ignore')
# 개행문자로 분리 (여러 바코드가 함께 들어온 경우)
lines = all_text.strip().split('\n')
self.raw_data_received.emit(f' → 분리된 라인 수: {len(lines)}')
for line in lines:
barcode_str = line.strip()
if not barcode_str:
continue
# 즉시 GUI에 표시
self.raw_data_received.emit(f' → 처리: "{barcode_str}" (길이: {len(barcode_str)})')
# 바코드 길이 검증 (13자리 EAN-13, 16자리 GS1-128만 허용)
valid_lengths = [13, 15, 16] # EAN-13, GS1-128 (01+13), GS1-128 (01+14)
if len(barcode_str) not in valid_lengths:
# 비정상 길이: 무시
self.raw_data_received.emit(f' → [무시] 비정상 길이 {len(barcode_str)}')
continue
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 바코드 데이터만 메인 스레드로 전달
self.raw_data_received.emit(f' → [OK] 시그널 전송!')
self.barcode_received.emit(barcode_str, timestamp, barcode_str.encode('utf-8'))
# 처리 완료 후 버퍼 확인
remaining = self.serial_connection.in_waiting
if remaining > 0:
self.raw_data_received.emit(f' → [주의] 처리 완료 후 버퍼에 {remaining} bytes 남음 (다음 루프에서 처리)')
except serial.SerialException as e:
self.connection_status.emit(False, f'포트 연결 실패: {str(e)}')
except Exception as e:
self.connection_status.emit(False, f'오류 발생: {str(e)}')
finally:
if self.serial_connection and self.serial_connection.is_open:
self.serial_connection.close()
def stop(self):
"""스레드 중지"""
self.running = False
if self.serial_connection and self.serial_connection.is_open:
self.serial_connection.close()
class BarcodeReaderGUI(QMainWindow):
"""바코드 리더 GUI 메인 윈도우"""
def __init__(self):
super().__init__()
self.reader_thread = None
self.scan_count = 0
self.search_threads = [] # 약품 조회 스레드 목록
self.generator_threads = [] # 라벨 생성 스레드 목록
self.init_ui()
def init_ui(self):
"""UI 초기화"""
self.setWindowTitle('허니웰 바코드 리더 - COM 포트')
self.setGeometry(100, 100, 900, 700)
# 중앙 위젯
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 메인 레이아웃
main_layout = QVBoxLayout()
central_widget.setLayout(main_layout)
# === 1. 설정 그룹 ===
settings_group = QGroupBox('연결 설정')
settings_layout = QHBoxLayout()
settings_group.setLayout(settings_layout)
# COM 포트 선택
settings_layout.addWidget(QLabel('COM 포트:'))
self.port_combo = QComboBox()
self.refresh_ports()
settings_layout.addWidget(self.port_combo)
# 새로고침 버튼
refresh_btn = QPushButton('새로고침')
refresh_btn.clicked.connect(self.refresh_ports)
settings_layout.addWidget(refresh_btn)
# 통신 속도
settings_layout.addWidget(QLabel('속도 (bps):'))
self.baudrate_spin = QSpinBox()
self.baudrate_spin.setMinimum(9600)
self.baudrate_spin.setMaximum(921600)
self.baudrate_spin.setValue(115200)
self.baudrate_spin.setSingleStep(9600)
settings_layout.addWidget(self.baudrate_spin)
# 수직 구분선
settings_layout.addWidget(QLabel('|'))
# 미리보기 모드 토글
self.preview_mode_checkbox = QCheckBox('미리보기 모드 (인쇄 안 함)')
self.preview_mode_checkbox.setChecked(True) # 기본값: 미리보기 (종이 절약!)
self.preview_mode_checkbox.setStyleSheet('font-size: 14px; color: #4CAF50; font-weight: bold;')
settings_layout.addWidget(self.preview_mode_checkbox)
settings_layout.addStretch()
main_layout.addWidget(settings_group)
# === 2. 제어 버튼 ===
control_layout = QHBoxLayout()
self.start_btn = QPushButton('시작')
self.start_btn.setStyleSheet('background-color: #4CAF50; color: white; font-weight: bold; padding: 10px;')
self.start_btn.clicked.connect(self.start_reading)
control_layout.addWidget(self.start_btn)
self.stop_btn = QPushButton('중지')
self.stop_btn.setStyleSheet('background-color: #f44336; color: white; font-weight: bold; padding: 10px;')
self.stop_btn.setEnabled(False)
self.stop_btn.clicked.connect(self.stop_reading)
control_layout.addWidget(self.stop_btn)
self.clear_btn = QPushButton('화면 지우기')
self.clear_btn.setStyleSheet('background-color: #2196F3; color: white; font-weight: bold; padding: 10px;')
self.clear_btn.clicked.connect(self.clear_output)
control_layout.addWidget(self.clear_btn)
main_layout.addLayout(control_layout)
# === 3. 상태 표시 ===
status_group = QGroupBox('상태')
status_layout = QVBoxLayout()
status_group.setLayout(status_layout)
self.status_label = QLabel('대기 중...')
self.status_label.setStyleSheet('color: gray; font-size: 14px; padding: 5px;')
status_layout.addWidget(self.status_label)
self.scan_count_label = QLabel('스캔 횟수: 0')
self.scan_count_label.setStyleSheet('color: blue; font-size: 14px; font-weight: bold; padding: 5px;')
status_layout.addWidget(self.scan_count_label)
main_layout.addWidget(status_group)
# === 4. 바코드 출력 영역 ===
output_group = QGroupBox('바코드 스캔 결과')
output_layout = QVBoxLayout()
output_group.setLayout(output_layout)
self.output_text = QTextEdit()
self.output_text.setReadOnly(True)
self.output_text.setFont(QFont('Consolas', 10))
self.output_text.setStyleSheet('background-color: #f5f5f5;')
output_layout.addWidget(self.output_text)
main_layout.addWidget(output_group)
def refresh_ports(self):
"""사용 가능한 COM 포트 새로고침"""
self.port_combo.clear()
ports = serial.tools.list_ports.comports()
for port in ports:
self.port_combo.addItem(f'{port.device} - {port.description}', port.device)
# COM3이 있으면 선택
for i in range(self.port_combo.count()):
if 'COM3' in self.port_combo.itemData(i):
self.port_combo.setCurrentIndex(i)
break
def start_reading(self):
"""바코드 읽기 시작"""
if self.reader_thread and self.reader_thread.isRunning():
return
# 선택된 포트와 속도 가져오기
port = self.port_combo.currentData()
if not port:
self.append_output('[오류] COM 포트를 선택해주세요.')
return
baudrate = self.baudrate_spin.value()
# 스레드 시작
self.reader_thread = BarcodeReaderThread(port, baudrate)
self.reader_thread.barcode_received.connect(self.on_barcode_received)
self.reader_thread.connection_status.connect(self.on_connection_status)
self.reader_thread.raw_data_received.connect(self.on_raw_data) # RAW 데이터 표시
self.reader_thread.start()
# UI 업데이트
self.start_btn.setEnabled(False)
self.stop_btn.setEnabled(True)
self.port_combo.setEnabled(False)
self.baudrate_spin.setEnabled(False)
self.status_label.setText(f'연결 시도 중... ({port}, {baudrate} bps)')
self.status_label.setStyleSheet('color: orange; font-size: 14px; padding: 5px;')
def stop_reading(self):
"""바코드 읽기 중지"""
if self.reader_thread:
self.reader_thread.stop()
self.reader_thread.wait()
# UI 업데이트
self.start_btn.setEnabled(True)
self.stop_btn.setEnabled(False)
self.port_combo.setEnabled(True)
self.baudrate_spin.setEnabled(True)
self.status_label.setText('중지됨')
self.status_label.setStyleSheet('color: gray; font-size: 14px; padding: 5px;')
self.append_output('[시스템] 바코드 리더를 중지했습니다.\n')
def on_connection_status(self, success, message):
"""연결 상태 업데이트"""
if success:
self.status_label.setText(f'연결됨: {message}')
self.status_label.setStyleSheet('color: green; font-size: 14px; font-weight: bold; padding: 5px;')
self.append_output(f'[시스템] {message}\n')
self.append_output('[대기] 바코드를 스캔해주세요...\n')
else:
self.status_label.setText(f'오류: {message}')
self.status_label.setStyleSheet('color: red; font-size: 14px; font-weight: bold; padding: 5px;')
self.append_output(f'[오류] {message}\n')
self.stop_reading()
def on_raw_data(self, log_message):
"""시리얼 포트 RAW 데이터 즉시 표시 (디버깅용)"""
self.append_output(log_message + '\n')
def on_barcode_received(self, barcode, timestamp, raw_data):
"""바코드 수신 처리 (DB 조회는 백그라운드 스레드로)"""
self.scan_count += 1
self.scan_count_label.setText(f'스캔 횟수: {self.scan_count}')
# 즉시 로그 출력 (DB 조회 전)
output = f'{"=" * 80}\n'
output += f'[스캔 #{self.scan_count}] {timestamp}\n'
output += f'바코드: {barcode}\n'
output += f'길이: {len(barcode)}\n'
output += f'[조회 중...] 약품 정보 검색 중\n'
self.append_output(output)
# 백그라운드 스레드로 DB 조회 작업 위임
search_thread = DrugSearchThread(barcode, timestamp, raw_data)
search_thread.search_complete.connect(self.on_search_complete)
search_thread.start()
self.search_threads.append(search_thread)
def on_search_complete(self, barcode, timestamp, raw_data, drug_info, parse_info):
"""약품 조회 완료 시그널 핸들러 (백그라운드 스레드에서 호출)"""
# 출력
output = ''
# GS1 파싱 정보 출력
if parse_info and parse_info.get('is_gs1'):
output += f'[GS1-128 바코드 감지]\n'
output += f' 원본 바코드: {parse_info["original"]}\n'
if parse_info.get('matched_barcode'):
output += f' 매칭된 바코드: {parse_info["matched_barcode"]}\n'
if len(parse_info.get('candidates', [])) > 1:
output += f' 검색 시도: {", ".join(parse_info["candidates"])}\n'
output += '\n'
# 약품 정보 출력
if drug_info:
output += f'[약품 정보]\n'
output += f' 약품명: {drug_info["goods_name"]}\n'
output += f' 약품코드: {drug_info["drug_code"]}\n'
output += f' 제조사: {drug_info["manufacturer"]}\n'
output += f' 매입가: {drug_info["price"]:,.0f}\n'
output += f' 판매가: {drug_info["sale_price"]:,.0f}\n'
if drug_info["sung_code"]:
output += f' 성분코드: {drug_info["sung_code"]}\n'
# 라벨 출력 또는 미리보기 (백그라운드 스레드)
try:
is_preview = self.preview_mode_checkbox.isChecked()
# 백그라운드 스레드로 이미지 생성 작업 위임
generator_thread = LabelGeneratorThread(
drug_info["goods_name"],
drug_info["sale_price"],
preview_mode=is_preview
)
# 완료 시그널 연결
generator_thread.image_ready.connect(self.on_label_generated)
# 스레드 시작 및 목록에 추가
generator_thread.start()
self.generator_threads.append(generator_thread)
# 로그 출력
if is_preview:
output += f'\n[미리보기] 이미지 생성 중...\n'
else:
output += f'\n[출력] 라벨 출력 중...\n'
except Exception as e:
output += f'\n[출력 오류] {str(e)}\n'
else:
output += f'[약품 정보] 데이터베이스에서 찾을 수 없습니다.\n'
output += f'\n원본(HEX): {raw_data.hex()}\n'
output += f'{"-" * 80}\n\n'
self.append_output(output)
# 완료된 스레드 정리
sender_thread = self.sender()
if sender_thread in self.search_threads:
self.search_threads.remove(sender_thread)
def on_label_generated(self, success, image_path, goods_name, error_msg):
"""
라벨 생성 완료 시그널 핸들러 (백그라운드 스레드에서 호출)
Args:
success: 성공 여부
image_path: 미리보기 이미지 경로 (미리보기 모드일 때만)
goods_name: 약품명
error_msg: 에러 메시지 (실패 시)
"""
if success:
if image_path:
# 미리보기 모드: Dialog 표시
self.append_output(f'[미리보기 완료] {goods_name}\n')
preview_dialog = LabelPreviewDialog(image_path, goods_name, self)
preview_dialog.show()
else:
# 실제 인쇄 모드: 성공 로그
self.append_output(f'[출력 완료] {goods_name} (192.168.0.168)\n')
else:
# 실패
self.append_output(f'[오류] {goods_name}: {error_msg}\n')
# 완료된 스레드 정리
sender_thread = self.sender()
if sender_thread in self.generator_threads:
self.generator_threads.remove(sender_thread)
def append_output(self, text):
"""출력 영역에 텍스트 추가"""
self.output_text.append(text)
# 스크롤을 맨 아래로
self.output_text.moveCursor(QTextCursor.End)
def clear_output(self):
"""출력 화면 지우기"""
self.output_text.clear()
self.scan_count = 0
self.scan_count_label.setText('스캔 횟수: 0')
def closeEvent(self, event):
"""프로그램 종료 시 스레드 정리"""
# 바코드 리더 스레드 종료
if self.reader_thread:
self.reader_thread.stop()
self.reader_thread.wait()
# 활성 약품 조회 스레드 종료
for thread in self.search_threads:
if thread.isRunning():
thread.wait()
# 활성 라벨 생성 스레드 종료
for thread in self.generator_threads:
if thread.isRunning():
thread.wait()
event.accept()
def main():
"""메인 함수"""
app = QApplication(sys.argv)
# 애플리케이션 스타일
app.setStyle('Fusion')
# 메인 윈도우 생성 및 표시
window = BarcodeReaderGUI()
window.show()
sys.exit(app.exec_())
if __name__ == '__main__':
main()