""" 허니웰 바코드 리더기 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()